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  * Write the current index information list to disk.
 797  */
 798 static void
 799 write_index_list ()
 800 {
 801   struct GNUNET_BIO_WriteHandle *wh;
 802   char *fn;
 803   struct IndexInfo *pos;
 804
 805   if (GNUNET_OK !=
 806       GNUNET_CONFIGURATION_get_value_filename (cfg,
 807                                                "FS",
 808                                                "INDEXDB",
 809                                                &fn))
 810     {
 811       GNUNET_log (GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
 812                   _("Configuration option `%s' in section `%s' missing.\n"),
 813                   "INDEXDB",
 814                   "FS");
 815       return;
 816     }
 817   wh = GNUNET_BIO_write_open (fn);
 818   if (NULL == wh)
 819     {
 820       GNUNET_log (GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
 821                   _("Could not open `%s'.\n"),
 822                   fn);
 823       GNUNET_free (fn);
 824       return;
 825     }
 826   pos = indexed_files;
 827   while (pos != NULL)
 828     {
 829       if ( (GNUNET_OK !=
 830             GNUNET_BIO_write (wh,
 831                               &pos->file_id,
 832                               sizeof (GNUNET_HashCode))) ||
 833            (GNUNET_OK !=
 834             GNUNET_BIO_write_string (wh,
 835                                      pos->filename)) )
 836         break;
 837       pos = pos->next;
 838     }
 839   if (GNUNET_OK !=
 840       GNUNET_BIO_write_close (wh))
 841     {
 842       GNUNET_log (GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
 843                   _("Error writing `%s'.\n"),
 844                   fn);
 845       GNUNET_free (fn);
 846       return;
 847     }
 848   GNUNET_free (fn);
 849 }
 850
 851
 852 /**
 853  * Read index information from disk.
 854  */
 855 static void
 856 read_index_list ()
 857 {
 858   struct GNUNET_BIO_ReadHandle *rh;
 859   char *fn;
 860   struct IndexInfo *pos;
 861   char *fname;
 862   GNUNET_HashCode hc;
 863   size_t slen;
 864   char *emsg;
 865
 866   if (GNUNET_OK !=
 867       GNUNET_CONFIGURATION_get_value_filename (cfg,
 868                                                "FS",
 869                                                "INDEXDB",
 870                                                &fn))
 871     {
 872       GNUNET_log (GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
 873                   _("Configuration option `%s' in section `%s' missing.\n"),
 874                   "INDEXDB",
 875                   "FS");
 876       return;
 877     }
 878   if (GNUNET_NO == GNUNET_DISK_file_test (fn))
 879     {
 880       /* no index info  yet */
 881       GNUNET_free (fn);
 882       return;
 883     }
 884   rh = GNUNET_BIO_read_open (fn);
 885   if (NULL == rh)
 886     {
 887       GNUNET_log (GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
 888                   _("Could not open `%s'.\n"),
 889                   fn);
 890       GNUNET_free (fn);
 891       return;
 892     }
 893
 894   while ( (GNUNET_OK ==
 895            GNUNET_BIO_read (rh,
 896                             "Hash of indexed file",
 897                             &hc,
 898                             sizeof (GNUNET_HashCode))) &&
 899           (GNUNET_OK ==
 900            GNUNET_BIO_read_string (rh,
 901                                    "Name of indexed file",
 902                                    &fname,
 903                                    1024 * 16)) )
 904     {
 905       slen = strlen (fname) + 1;
 906       pos = GNUNET_malloc (sizeof (struct IndexInfo) + slen);
 907       pos->file_id = hc;
 908       pos->filename = (const char *) &pos[1];
 909       memcpy (&pos[1], fname, slen);
 910       if (GNUNET_SYSERR ==
 911           GNUNET_CONTAINER_multihashmap_put (ifm,
 912                                              &hc,
 913                                              (void*) pos->filename,
 914                                              GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY))
 915         {
 916           GNUNET_free (pos);
 917         }
 918       else
 919         {
 920           pos->next = indexed_files;
 921           indexed_files = pos;
 922         }
 923       GNUNET_free (fname);
 924     }
 925   if (GNUNET_OK !=
 926       GNUNET_BIO_read_close (rh, &emsg))
 927     GNUNET_free (emsg);
 928   GNUNET_free (fn);
 929 }
 930
 931
 932 /**
 933  * We've validated the hash of the file we're about to
 934  * index.  Signal success to the client and update
 935  * our internal data structures.
 936  *
 937  * @param ii the index info entry for the request
 938  */
 939 static void
 940 signal_index_ok (struct IndexInfo *ii)
 941 {
 942   if (GNUNET_SYSERR ==
 943       GNUNET_CONTAINER_multihashmap_put (ifm,
 944                                          &ii->file_id,
 945                                          (void*) ii->filename,
 946                                          GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY))
 947     {
 948       GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
 949                   _("Index request received for file `%s' is indexed as `%s'.  Permitting anyway.\n"),
 950                   ii->filename,
 951                   (const char*) GNUNET_CONTAINER_multihashmap_get (ifm,
 952                                                                    &ii->file_id));
 953       GNUNET_SERVER_transmit_context_append (ii->tc,
 954                                              NULL, 0,
 955                                              GNUNET_MESSAGE_TYPE_FS_INDEX_START_OK);
 956       GNUNET_SERVER_transmit_context_run (ii->tc,
 957                                           GNUNET_TIME_UNIT_MINUTES);
 958       GNUNET_free (ii);
 959       return;
 960     }
 961   ii->next = indexed_files;
 962   indexed_files = ii;
 963   write_index_list ();
 964   GNUNET_SERVER_transmit_context_append (ii->tc,
 965                                          NULL, 0,
 966                                          GNUNET_MESSAGE_TYPE_FS_INDEX_START_OK);
 967   GNUNET_SERVER_transmit_context_run (ii->tc,
 968                                       GNUNET_TIME_UNIT_MINUTES);
 969   ii->tc = NULL;
 970 }
 971
 972
 973 /**
 974  * Function called once the hash computation over an
 975  * indexed file has completed.
 976  *
 977  * @param cls closure, our publishing context
 978  * @param res resulting hash, NULL on error
 979  */
 980 static void
 981 hash_for_index_val (void *cls,
 982                     const GNUNET_HashCode *
 983                     res)
 984 {
 985   struct IndexInfo *ii = cls;
 986
 987   if ( (res == NULL) ||
 988        (0 != memcmp (res,
 989                      &ii->file_id,
 990                      sizeof(GNUNET_HashCode))) )
 991     {
 992       GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
 993                   _("Hash mismatch trying to index file `%s' which has hash `%s'\n"),
 994                   ii->filename,
 995                   GNUNET_h2s (res));
 996 #if DEBUG_FS
 997       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 998                   "Wanted `%s'\n",
 999                   GNUNET_h2s (&ii->file_id));
1000 #endif
1001       GNUNET_SERVER_transmit_context_append (ii->tc,
1002                                              NULL, 0,
1003                                              GNUNET_MESSAGE_TYPE_FS_INDEX_START_FAILED);
1004       GNUNET_SERVER_transmit_context_run (ii->tc,
1005                                           GNUNET_TIME_UNIT_MINUTES);
1006       GNUNET_free (ii);
1007       return;
1008     }
1009   signal_index_ok (ii);
1010 }
1011
1012
1013 /**
1014  * Handle INDEX_START-message.
1015  *
1016  * @param cls closure
1017  * @param client identification of the client
1018  * @param message the actual message
1019  */
1020 static void
1021 handle_index_start (void *cls,
1022                     struct GNUNET_SERVER_Client *client,
1023                     const struct GNUNET_MessageHeader *message)
1024 {
1025   const struct IndexStartMessage *ism;
1026   const char *fn;
1027   uint16_t msize;
1028   struct IndexInfo *ii;
1029   size_t slen;
1030   uint32_t dev;
1031   uint64_t ino;
1032   uint32_t mydev;
1033   uint64_t myino;
1034
1035   msize = ntohs(message->size);
1036   if ( (msize <= sizeof (struct IndexStartMessage)) ||
1037        ( ((const char *)message)[msize-1] != '\0') )
1038     {
1039       GNUNET_break (0);
1040       GNUNET_SERVER_receive_done (client,
1041                                   GNUNET_SYSERR);
1042       return;
1043     }
1044   ism = (const struct IndexStartMessage*) message;
1045   fn = (const char*) &ism[1];
1046   dev = ntohl (ism->device);
1047   ino = GNUNET_ntohll (ism->inode);
1048   ism = (const struct IndexStartMessage*) message;
1049   slen = strlen (fn) + 1;
1050   ii = GNUNET_malloc (sizeof (struct IndexInfo) + slen);
1051   ii->filename = (const char*) &ii[1];
1052   memcpy (&ii[1], fn, slen);
1053   ii->file_id = ism->file_id;
1054   ii->tc = GNUNET_SERVER_transmit_context_create (client);
1055   if ( ( (dev != 0) ||
1056          (ino != 0) ) &&
1057        (GNUNET_OK == GNUNET_DISK_file_get_identifiers (fn,
1058                                                        &mydev,
1059                                                        &myino)) &&
1060        ( (dev == mydev) &&
1061          (ino == myino) ) )
1062     {
1063       /* fast validation OK! */
1064       signal_index_ok (ii);
1065       return;
1066     }
1067 #if DEBUG_FS
1068   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1069               "Mismatch in file identifiers (%llu != %llu or %u != %u), need to hash.\n",
1070               (unsigned long long) ino,
1071               (unsigned long long) myino,
1072               (unsigned int) dev,
1073               (unsigned int) mydev);
1074 #endif
1075   /* slow validation, need to hash full file (again) */
1076   GNUNET_CRYPTO_hash_file (sched,
1077                            GNUNET_SCHEDULER_PRIORITY_IDLE,
1078                            fn,
1079                            HASHING_BLOCKSIZE,
1080                            &hash_for_index_val,
1081                            ii);
1082 }
1083
1084
1085 /**
1086  * Handle INDEX_LIST_GET-message.
1087  *
1088  * @param cls closure
1089  * @param client identification of the client
1090  * @param message the actual message
1091  */
1092 static void
1093 handle_index_list_get (void *cls,
1094                        struct GNUNET_SERVER_Client *client,
1095                        const struct GNUNET_MessageHeader *message)
1096 {
1097   struct GNUNET_SERVER_TransmitContext *tc;
1098   struct IndexInfoMessage *iim;
1099   char buf[GNUNET_SERVER_MAX_MESSAGE_SIZE];
1100   size_t slen;
1101   const char *fn;
1102   struct GNUNET_MessageHeader *msg;
1103   struct IndexInfo *pos;
1104
1105   tc = GNUNET_SERVER_transmit_context_create (client);
1106   iim = (struct IndexInfoMessage*) buf;
1107   msg = &iim->header;
1108   pos = indexed_files;
1109   while (NULL != pos)
1110     {
1111       iim->reserved = 0;
1112       iim->file_id = pos->file_id;
1113       fn = pos->filename;
1114       slen = strlen (fn) + 1;
1115       if (slen + sizeof (struct IndexInfoMessage) >
1116           GNUNET_SERVER_MAX_MESSAGE_SIZE)
1117         {
1118           GNUNET_break (0);
1119           break;
1120         }
1121       memcpy (&iim[1], fn, slen);
1122       GNUNET_SERVER_transmit_context_append
1123         (tc,
1124          &msg[1],
1125          sizeof (struct IndexInfoMessage)
1126          - sizeof (struct GNUNET_MessageHeader) + slen,
1127          GNUNET_MESSAGE_TYPE_FS_INDEX_LIST_ENTRY);
1128       pos = pos->next;
1129     }
1130   GNUNET_SERVER_transmit_context_append (tc,
1131                                          NULL, 0,
1132                                          GNUNET_MESSAGE_TYPE_FS_INDEX_LIST_END);
1133   GNUNET_SERVER_transmit_context_run (tc,
1134                                       GNUNET_TIME_UNIT_MINUTES);
1135 }
1136
1137
1138 /**
1139  * Handle UNINDEX-message.
1140  *
1141  * @param cls closure
1142  * @param client identification of the client
1143  * @param message the actual message
1144  */
1145 static void
1146 handle_unindex (void *cls,
1147                 struct GNUNET_SERVER_Client *client,
1148                 const struct GNUNET_MessageHeader *message)
1149 {
1150   const struct UnindexMessage *um;
1151   struct IndexInfo *pos;
1152   struct IndexInfo *prev;
1153   struct IndexInfo *next;
1154   struct GNUNET_SERVER_TransmitContext *tc;
1155   int found;
1156
1157   um = (const struct UnindexMessage*) message;
1158   found = GNUNET_NO;
1159   prev = NULL;
1160   pos = indexed_files;
1161   while (NULL != pos)
1162     {
1163       next = pos->next;
1164       if (0 == memcmp (&pos->file_id,
1165                        &um->file_id,
1166                        sizeof (GNUNET_HashCode)))
1167         {
1168           if (prev == NULL)
1169             indexed_files = pos->next;
1170           else
1171             prev->next = pos->next;
1172           GNUNET_free (pos);
1173           found = GNUNET_YES;
1174         }
1175       else
1176         {
1177           prev = pos;
1178         }
1179       pos = next;
1180     }
1181 #if DEBUG_FS
1182   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1183               "Client requested unindexing of file `%s': %s\n",
1184               GNUNET_h2s (&um->file_id),
1185               found ? "found" : "not found");
1186 #endif
1187   if (GNUNET_YES == found)
1188     write_index_list ();
1189   tc = GNUNET_SERVER_transmit_context_create (client);
1190   GNUNET_SERVER_transmit_context_append (tc,
1191                                          NULL, 0,
1192                                          GNUNET_MESSAGE_TYPE_FS_UNINDEX_OK);
1193   GNUNET_SERVER_transmit_context_run (tc,
1194                                       GNUNET_TIME_UNIT_MINUTES);
1195 }
1196
1197
1198 /**
1199  * Run the next DS request in our
1200  * queue, we're done with the current one.
1201  */
1202 static void
1203 next_ds_request ()
1204 {
1205   struct DatastoreRequestQueue *e;
1206
1207   while (NULL != (e = drq_head))
1208     {
1209       if (0 != GNUNET_TIME_absolute_get_remaining (e->timeout).value)
1210         break;
1211       if (e->task != GNUNET_SCHEDULER_NO_TASK)
1212         GNUNET_SCHEDULER_cancel (sched, e->task);
1213       GNUNET_CONTAINER_DLL_remove (drq_head, drq_tail, e);
1214       e->req (e->req_cls, GNUNET_NO);
1215       GNUNET_free (e);
1216     }
1217   if (e == NULL)
1218     return;
1219   if (e->task != GNUNET_SCHEDULER_NO_TASK)
1220     GNUNET_SCHEDULER_cancel (sched, e->task);
1221   e->task = GNUNET_SCHEDULER_NO_TASK;
1222   e->req (e->req_cls, GNUNET_YES);
1223   GNUNET_CONTAINER_DLL_remove (drq_head, drq_tail, e);
1224   GNUNET_free (e);
1225 }
1226
1227
1228 /**
1229  * A datastore request had to be timed out.
1230  *
1231  * @param cls closure (of type "struct DatastoreRequestQueue*")
1232  * @param tc task context, unused
1233  */
1234 static void
1235 timeout_ds_request (void *cls,
1236                     const struct GNUNET_SCHEDULER_TaskContext *tc)
1237 {
1238   struct DatastoreRequestQueue *e = cls;
1239
1240   e->task = GNUNET_SCHEDULER_NO_TASK;
1241   GNUNET_CONTAINER_DLL_remove (drq_head, drq_tail, e);
1242   e->req (e->req_cls, GNUNET_NO);
1243   GNUNET_free (e);
1244 }
1245
1246
1247 /**
1248  * Queue a request for the datastore.
1249  *
1250  * @param deadline by when the request should run
1251  * @param fun function to call once the request can be run
1252  * @param fun_cls closure for fun
1253  */
1254 static struct DatastoreRequestQueue *
1255 queue_ds_request (struct GNUNET_TIME_Relative deadline,
1256                   RequestFunction fun,
1257                   void *fun_cls)
1258 {
1259   struct DatastoreRequestQueue *e;
1260   struct DatastoreRequestQueue *bef;
1261
1262   if (drq_head == NULL)
1263     {
1264       /* no other requests pending, run immediately */
1265       fun (fun_cls, GNUNET_OK);
1266       return NULL;
1267     }
1268   e = GNUNET_malloc (sizeof (struct DatastoreRequestQueue));
1269   e->timeout = GNUNET_TIME_relative_to_absolute (deadline);
1270   e->req = fun;
1271   e->req_cls = fun_cls;
1272   if (deadline.value == GNUNET_TIME_UNIT_FOREVER_REL.value)
1273     {
1274       /* local request, highest prio, put at head of queue
1275          regardless of deadline */
1276       bef = NULL;
1277     }
1278   else
1279     {
1280       bef = drq_tail;
1281       while ( (NULL != bef) &&
1282               (e->timeout.value < bef->timeout.value) )
1283         bef = bef->prev;
1284     }
1285   GNUNET_CONTAINER_DLL_insert_after (drq_head, drq_tail, bef, e);
1286   if (deadline.value == GNUNET_TIME_UNIT_FOREVER_REL.value)
1287     return e;
1288   e->task = GNUNET_SCHEDULER_add_delayed (sched,
1289                                           deadline,
1290                                           &timeout_ds_request,
1291                                           e);
1292   return e;
1293 }
1294
1295
1296 /**
1297  * Free the state associated with a local get context.
1298  *
1299  * @param lgc the lgc to free
1300  */
1301 static void
1302 local_get_context_free (struct LocalGetContext *lgc)
1303 {
1304   GNUNET_CONTAINER_DLL_remove (lgc_head, lgc_tail, lgc);
1305   GNUNET_SERVER_client_drop (lgc->client);
1306   GNUNET_free_non_null (lgc->results);
1307   if (lgc->results_bf != NULL)
1308     GNUNET_CONTAINER_bloomfilter_free (lgc->results_bf);
1309   if (lgc->req != NULL)
1310     {
1311       if (lgc->req->task != GNUNET_SCHEDULER_NO_TASK)
1312         GNUNET_SCHEDULER_cancel (sched, lgc->req->task);
1313       GNUNET_CONTAINER_DLL_remove (lgc_head, lgc_tail, lgc);
1314       GNUNET_free (lgc->req);
1315     }
1316   GNUNET_free (lgc);
1317 }
1318
1319
1320 /**
1321  * We're able to transmit the next (local) result to the client.
1322  * Do it and ask the datastore for more.  Or, on error, tell
1323  * the datastore to stop giving us more.
1324  *
1325  * @param cls our closure (struct LocalGetContext)
1326  * @param max maximum number of bytes we can transmit
1327  * @param buf where to copy our message
1328  * @return number of bytes copied to buf
1329  */
1330 static size_t
1331 transmit_local_result (void *cls,
1332                        size_t max,
1333                        void *buf)
1334 {
1335   struct LocalGetContext *lgc = cls;
1336   uint16_t msize;
1337
1338   if (NULL == buf)
1339     {
1340 #if DEBUG_FS
1341       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1342                   "Failed to transmit result to local client, aborting datastore iteration.\n");
1343 #endif
1344       /* error, abort! */
1345       GNUNET_free (lgc->result);
1346       lgc->result = NULL;
1347       GNUNET_DATASTORE_get_next (dsh, GNUNET_NO);
1348       return 0;
1349     }
1350   msize = ntohs (lgc->result->header.size);
1351 #if DEBUG_FS
1352   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1353               "Transmitting %u bytes of result to local client.\n",
1354               msize);
1355 #endif
1356   GNUNET_assert (max >= msize);
1357   memcpy (buf, lgc->result, msize);
1358   GNUNET_free (lgc->result);
1359   lgc->result = NULL;
1360   GNUNET_DATASTORE_get_next (dsh, GNUNET_YES);
1361   return msize;
1362 }
1363
1364
1365 /**
1366  * Continuation called from datastore's remove
1367  * function.
1368  *
1369  * @param cls unused
1370  * @param success did the deletion work?
1371  * @param msg error message
1372  */
1373 static void
1374 remove_cont (void *cls,
1375              int success,
1376              const char *msg)
1377 {
1378   if (GNUNET_OK != success)
1379     GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
1380                 _("Failed to delete bogus block: %s\n"),
1381                 msg);
1382   GNUNET_DATASTORE_get_next (dsh, GNUNET_YES);
1383 }
1384
1385
1386 /**
1387  * Mingle hash with the mingle_number to
1388  * produce different bits.
1389  */
1390 static void
1391 mingle_hash (const GNUNET_HashCode * in,
1392              int32_t mingle_number,
1393              GNUNET_HashCode * hc)
1394 {
1395   GNUNET_HashCode m;
1396
1397   GNUNET_CRYPTO_hash (&mingle_number,
1398                       sizeof (int32_t),
1399                       &m);
1400   GNUNET_CRYPTO_hash_xor (&m, in, hc);
1401 }
1402
1403
1404 /**
1405  * We've received an on-demand encoded block
1406  * from the datastore.  Attempt to do on-demand
1407  * encoding and (if successful), call the
1408  * continuation with the resulting block.  On
1409  * error, clean up and ask the datastore for
1410  * more results.
1411  *
1412  * @param key key for the content
1413  * @param size number of bytes in data
1414  * @param data content stored
1415  * @param type type of the content
1416  * @param priority priority of the content
1417  * @param anonymity anonymity-level for the content
1418  * @param expiration expiration time for the content
1419  * @param uid unique identifier for the datum;
1420  *        maybe 0 if no unique identifier is available
1421  * @param cont function to call with the actual block
1422  * @param cont_cls closure for cont
1423  */
1424 static void
1425 handle_on_demand_block (const GNUNET_HashCode * key,
1426                         uint32_t size,
1427                         const void *data,
1428                         uint32_t type,
1429                         uint32_t priority,
1430                         uint32_t anonymity,
1431                         struct GNUNET_TIME_Absolute
1432                         expiration, uint64_t uid,
1433                         GNUNET_DATASTORE_Iterator cont,
1434                         void *cont_cls)
1435 {
1436   const struct OnDemandBlock *odb;
1437   GNUNET_HashCode nkey;
1438   struct GNUNET_CRYPTO_AesSessionKey skey;
1439   struct GNUNET_CRYPTO_AesInitializationVector iv;
1440   GNUNET_HashCode query;
1441   ssize_t nsize;
1442   char ndata[DBLOCK_SIZE];
1443   char edata[DBLOCK_SIZE];
1444   const char *fn;
1445   struct GNUNET_DISK_FileHandle *fh;
1446   uint64_t off;
1447
1448   if (size != sizeof (struct OnDemandBlock))
1449     {
1450       GNUNET_break (0);
1451       GNUNET_DATASTORE_remove (dsh,
1452                                key,
1453                                size,
1454                                data,
1455                                &remove_cont,
1456                                NULL,
1457                                GNUNET_TIME_UNIT_FOREVER_REL);
1458       return;
1459     }
1460   odb = (const struct OnDemandBlock*) data;
1461   off = GNUNET_ntohll (odb->offset);
1462   fn = (const char*) GNUNET_CONTAINER_multihashmap_get (ifm,
1463                                                         &odb->file_id);
1464   fh = NULL;
1465   if ( (NULL == fn) ||
1466        (NULL == (fh = GNUNET_DISK_file_open (fn,
1467                                              GNUNET_DISK_OPEN_READ,
1468                                              GNUNET_DISK_PERM_NONE))) ||
1469        (off !=
1470         GNUNET_DISK_file_seek (fh,
1471                                off,
1472                                GNUNET_DISK_SEEK_SET)) ||
1473        (-1 ==
1474         (nsize = GNUNET_DISK_file_read (fh,
1475                                         ndata,
1476                                         sizeof (ndata)))) )
1477     {
1478       GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
1479                   _("Could not access indexed file `%s' at offset %llu: %s\n"),
1480                   GNUNET_h2s (&odb->file_id),
1481                   (unsigned long long) off,
1482                   STRERROR (errno));
1483       if (fh != NULL)
1484         GNUNET_DISK_file_close (fh);
1485       /* FIXME: if this happens often, we need
1486          to remove the OnDemand block from the DS! */
1487       GNUNET_DATASTORE_get_next (dsh, GNUNET_YES);
1488       return;
1489     }
1490   GNUNET_DISK_file_close (fh);
1491   GNUNET_CRYPTO_hash (ndata,
1492                       nsize,
1493                       &nkey);
1494   GNUNET_CRYPTO_hash_to_aes_key (&nkey, &skey, &iv);
1495   GNUNET_CRYPTO_aes_encrypt (ndata,
1496                              nsize,
1497                              &skey,
1498                              &iv,
1499                              edata);
1500   GNUNET_CRYPTO_hash (edata,
1501                       nsize,
1502                       &query);
1503   if (0 != memcmp (&query,
1504                    key,
1505                    sizeof (GNUNET_HashCode)))
1506     {
1507       GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
1508                   _("Indexed file `%s' changed at offset %llu\n"),
1509                   fn,
1510                   (unsigned long long) off);
1511       /* FIXME: if this happens often, we need
1512          to remove the OnDemand block from the DS! */
1513       GNUNET_DATASTORE_get_next (dsh, GNUNET_YES);
1514       return;
1515     }
1516   cont (cont_cls,
1517         key,
1518         nsize,
1519         edata,
1520         GNUNET_DATASTORE_BLOCKTYPE_DBLOCK,
1521         priority,
1522         anonymity,
1523         expiration,
1524         uid);
1525 }
1526
1527
1528 /**
1529  * How many bytes should a bloomfilter be if we have already seen
1530  * entry_count responses?  Note that BLOOMFILTER_K gives us the number
1531  * of bits set per entry.  Furthermore, we should not re-size the
1532  * filter too often (to keep it cheap).
1533  *
1534  * Since other peers will also add entries but not resize the filter,
1535  * we should generally pick a slightly larger size than what the
1536  * strict math would suggest.
1537  *
1538  * @return must be a power of two and smaller or equal to 2^15.
1539  */
1540 static size_t
1541 compute_bloomfilter_size (unsigned int entry_count)
1542 {
1543   size_t size;
1544   unsigned int ideal = (entry_count * BLOOMFILTER_K) / 4;
1545   uint16_t max = 1 << 15;
1546
1547   if (entry_count > max)
1548     return max;
1549   size = 8;
1550   while ((size < max) && (size < ideal))
1551     size *= 2;
1552   if (size > max)
1553     return max;
1554   return size;
1555 }
1556
1557
1558 /**
1559  * Recalculate our bloom filter for filtering replies.
1560  *
1561  * @param count number of entries we are filtering right now
1562  * @param mingle set to our new mingling value
1563  * @param bf_size set to the size of the bloomfilter
1564  * @param entries the entries to filter
1565  * @return updated bloomfilter, NULL for none
1566  */
1567 static struct GNUNET_CONTAINER_BloomFilter *
1568 refresh_bloomfilter (unsigned int count,
1569                      int32_t * mingle,
1570                      size_t *bf_size,
1571                      const GNUNET_HashCode *entries)
1572 {
1573   struct GNUNET_CONTAINER_BloomFilter *bf;
1574   size_t nsize;
1575   unsigned int i;
1576   GNUNET_HashCode mhash;
1577
1578   if (0 == count)
1579     return NULL;
1580   nsize = compute_bloomfilter_size (count);
1581   *mingle = GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK, -1);
1582   *bf_size = nsize;
1583   bf = GNUNET_CONTAINER_bloomfilter_init (NULL,
1584                                           nsize,
1585                                           BLOOMFILTER_K);
1586   for (i=0;i<count;i++)
1587     {
1588       mingle_hash (&entries[i], *mingle, &mhash);
1589       GNUNET_CONTAINER_bloomfilter_add (bf, &mhash);
1590     }
1591   return bf;
1592 }
1593
1594
1595 /**
1596  * Closure used for "target_peer_select_cb".
1597  */
1598 struct PeerSelectionContext
1599 {
1600   /**
1601    * The request for which we are selecting
1602    * peers.
1603    */
1604   struct PendingRequest *pr;
1605
1606   /**
1607    * Current "prime" target.
1608    */
1609   struct GNUNET_PeerIdentity target;
1610
1611   /**
1612    * How much do we like this target?
1613    */
1614   double target_score;
1615
1616 };
1617
1618
1619 /**
1620  * Function called for each connected peer to determine
1621  * which one(s) would make good targets for forwarding.
1622  *
1623  * @param cls closure (struct PeerSelectionContext)
1624  * @param key current key code (peer identity)
1625  * @param value value in the hash map (struct ConnectedPeer)
1626  * @return GNUNET_YES if we should continue to
1627  *         iterate,
1628  *         GNUNET_NO if not.
1629  */
1630 static int
1631 target_peer_select_cb (void *cls,
1632                        const GNUNET_HashCode * key,
1633                        void *value)
1634 {
1635   struct PeerSelectionContext *psc = cls;
1636   struct ConnectedPeer *cp = value;
1637   struct PendingRequest *pr = psc->pr;
1638   double score;
1639   unsigned int i;
1640
1641   /* 1) check if we have already (recently) forwarded to this peer */
1642   for (i=0;i<pr->used_pids_off;i++)
1643     if (pr->used_pids[i] == cp->pid)
1644       return GNUNET_YES; /* skip */
1645   // 2) calculate how much we'd like to forward to this peer
1646   score = 0; // FIXME!
1647
1648   /* store best-fit in closure */
1649   if (score > psc->target_score)
1650     {
1651       psc->target_score = score;
1652       psc->target.hashPubKey = *key;
1653     }
1654   return GNUNET_YES;
1655 }
1656
1657
1658 /**
1659  * We use a random delay to make the timing of requests
1660  * less predictable.  This function returns such a random
1661  * delay.
1662  *
1663  * @return random delay to use for some request, between 0 and TTL_DECREMENT ms
1664  */
1665 static struct GNUNET_TIME_Relative
1666 get_processing_delay ()
1667 {
1668   return GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MILLISECONDS,
1669                                         GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
1670                                                                   TTL_DECREMENT));
1671 }
1672
1673
1674 /**
1675  * Task that is run for each request with the
1676  * goal of forwarding the associated query to
1677  * other peers.  The task should re-schedule
1678  * itself to be re-run once the TTL has expired.
1679  * (or at a later time if more peers should
1680  * be queried earlier).
1681  *
1682  * @param cls the requests "struct PendingRequest*"
1683  * @param tc task context (unused)
1684  */
1685 static void
1686 forward_request_task (void *cls,
1687                       const struct GNUNET_SCHEDULER_TaskContext *tc);
1688
1689
1690 /**
1691  * We've selected a peer for forwarding of a query.
1692  * Construct the message and then re-schedule the
1693  * task to forward again to (other) peers.
1694  *
1695  * @param cls closure
1696  * @param size number of bytes available in buf
1697  * @param buf where the callee should write the message
1698  * @return number of bytes written to buf
1699  */
1700 static size_t
1701 transmit_request_cb (void *cls,
1702                      size_t size,
1703                      void *buf)
1704 {
1705   struct PendingRequest *pr = cls;
1706   struct GetMessage *gm;
1707   GNUNET_HashCode *ext;
1708   char *bfdata;
1709   size_t msize;
1710   unsigned int k;
1711
1712   pr->cth = NULL;
1713   /* (1) check for timeout */
1714   if (NULL == buf)
1715     {
1716       /* timeout, try another peer immediately again */
1717       pr->task = GNUNET_SCHEDULER_add_with_priority (sched,
1718                                                      GNUNET_SCHEDULER_PRIORITY_IDLE,
1719                                                      &forward_request_task,
1720                                                      pr);
1721       return 0;
1722     }
1723   /* (2) build query message */
1724   k = 0; // FIXME: count hash codes!
1725   msize = sizeof (struct GetMessage) + pr->bf_size + k * sizeof(GNUNET_HashCode);
1726   GNUNET_assert (msize < GNUNET_SERVER_MAX_MESSAGE_SIZE);
1727   gm = (struct GetMessage*) buf;
1728   gm->header.type = htons (GNUNET_MESSAGE_TYPE_FS_GET);
1729   gm->header.size = htons (msize);
1730   gm->type = htonl (pr->type);
1731   pr->remaining_priority /= 2;
1732   gm->priority = htonl (pr->remaining_priority);
1733   gm->ttl = htonl (pr->ttl);
1734   gm->filter_mutator = htonl(pr->mingle);
1735   gm->hash_bitmap = htonl (42);
1736   gm->query = pr->query;
1737   ext = (GNUNET_HashCode*) &gm[1];
1738   // FIXME: setup "ext[0]..[k-1]"
1739   bfdata = (char *) &ext[k];
1740   if (pr->bf != NULL)
1741     GNUNET_CONTAINER_bloomfilter_get_raw_data (pr->bf,
1742                                                bfdata,
1743                                                pr->bf_size);
1744
1745   /* (3) schedule job to do it again (or another peer, etc.) */
1746   pr->task = GNUNET_SCHEDULER_add_delayed (sched,
1747                                            get_processing_delay (), // FIXME!
1748                                            &forward_request_task,
1749                                            pr);
1750
1751   return msize;
1752 }
1753
1754
1755 /**
1756  * Function called after we've tried to reserve
1757  * a certain amount of bandwidth for a reply.
1758  * Check if we succeeded and if so send our query.
1759  *
1760  * @param cls the requests "struct PendingRequest*"
1761  * @param peer identifies the peer
1762  * @param latency current latency estimate, "FOREVER" if we have been
1763  *                disconnected
1764  * @param bpm_in set to the current bandwidth limit (receiving) for this peer
1765  * @param bpm_out set to the current bandwidth limit (sending) for this peer
1766  * @param amount set to the amount that was actually reserved or unreserved
1767  * @param preference current traffic preference for the given peer
1768  */
1769 static void
1770 target_reservation_cb (void *cls,
1771                        const struct
1772                        GNUNET_PeerIdentity * peer,
1773                        unsigned int bpm_in,
1774                        unsigned int bpm_out,
1775                        struct GNUNET_TIME_Relative
1776                        latency, int amount,
1777                        unsigned long long preference)
1778 {
1779   struct PendingRequest *pr = cls;
1780   uint32_t priority;
1781   uint16_t size;
1782   struct GNUNET_TIME_Relative maxdelay;
1783
1784   GNUNET_assert (peer != NULL);
1785   if ( (amount != DBLOCK_SIZE) ||
1786        (pr->cth != NULL) )
1787     {
1788       /* try again later; FIXME: we may need to un-reserve "amount"? */
1789       pr->task = GNUNET_SCHEDULER_add_delayed (sched,
1790                                                get_processing_delay (), // FIXME: longer?
1791                                                &forward_request_task,
1792                                                pr);
1793       return;
1794     }
1795   // (2) transmit, update ttl/priority
1796   // FIXME: calculate priority, maxdelay, size properly!
1797   priority = 0;
1798   size = 60000;
1799   maxdelay = GNUNET_CONSTANTS_SERVICE_TIMEOUT;
1800   pr->cth = GNUNET_CORE_notify_transmit_ready (core,
1801                                                priority,
1802                                                maxdelay,
1803                                                peer,
1804                                                size,
1805                                                &transmit_request_cb,
1806                                                pr);
1807   if (pr->cth == NULL)
1808     {
1809       /* try again later */
1810       pr->task = GNUNET_SCHEDULER_add_delayed (sched,
1811                                                get_processing_delay (), // FIXME: longer?
1812                                                &forward_request_task,
1813                                                pr);
1814     }
1815 }
1816
1817
1818 /**
1819  * Task that is run for each request with the
1820  * goal of forwarding the associated query to
1821  * other peers.  The task should re-schedule
1822  * itself to be re-run once the TTL has expired.
1823  * (or at a later time if more peers should
1824  * be queried earlier).
1825  *
1826  * @param cls the requests "struct PendingRequest*"
1827  * @param tc task context (unused)
1828  */
1829 static void
1830 forward_request_task (void *cls,
1831                       const struct GNUNET_SCHEDULER_TaskContext *tc)
1832 {
1833   struct PendingRequest *pr = cls;
1834   struct PeerSelectionContext psc;
1835
1836   pr->task = GNUNET_SCHEDULER_NO_TASK;
1837   if (pr->cth != NULL)
1838     {
1839       /* we're busy transmitting a result, wait a bit */
1840       pr->task = GNUNET_SCHEDULER_add_delayed (sched,
1841                                                get_processing_delay (),
1842                                                &forward_request_task,
1843                                                pr);
1844       return;
1845     }
1846   /* (1) select target */
1847   psc.pr = pr;
1848   psc.target_score = DBL_MIN;
1849   GNUNET_CONTAINER_multihashmap_iterate (connected_peers,
1850                                          &target_peer_select_cb,
1851                                          &psc);
1852   if (psc.target_score == DBL_MIN)
1853     {
1854       /* no possible target found, wait some time */
1855       pr->task = GNUNET_SCHEDULER_add_delayed (sched,
1856                                                get_processing_delay (), // FIXME: exponential back-off? or at least wait longer...
1857                                                &forward_request_task,
1858                                                pr);
1859       return;
1860     }
1861   /* (2) reserve reply bandwidth */
1862   // FIXME: need a way to cancel; this
1863   // async operation is problematic (segv-problematic)
1864   // if "pr" is destroyed while it happens!
1865   GNUNET_CORE_peer_configure (core,
1866                               &psc.target,
1867                               GNUNET_CONSTANTS_SERVICE_TIMEOUT,
1868                               -1,
1869                               DBLOCK_SIZE, // FIXME: make dependent on type?
1870                               0,
1871                               &target_reservation_cb,
1872                               pr);
1873 }
1874
1875
1876 /**
1877  * We're processing (local) results for a search request
1878  * from a (local) client.  Pass applicable results to the
1879  * client and if we are done either clean up (operation
1880  * complete) or switch to P2P search (more results possible).
1881  *
1882  * @param cls our closure (struct LocalGetContext)
1883  * @param key key for the content
1884  * @param size number of bytes in data
1885  * @param data content stored
1886  * @param type type of the content
1887  * @param priority priority of the content
1888  * @param anonymity anonymity-level for the content
1889  * @param expiration expiration time for the content
1890  * @param uid unique identifier for the datum;
1891  *        maybe 0 if no unique identifier is available
1892  */
1893 static void
1894 process_local_get_result (void *cls,
1895                           const GNUNET_HashCode * key,
1896                           uint32_t size,
1897                           const void *data,
1898                           uint32_t type,
1899                           uint32_t priority,
1900                           uint32_t anonymity,
1901                           struct GNUNET_TIME_Absolute
1902                           expiration,
1903                           uint64_t uid)
1904 {
1905   struct LocalGetContext *lgc = cls;
1906   struct PendingRequest *pr;
1907   struct ClientRequestList *crl;
1908   struct ClientList *cl;
1909   size_t msize;
1910   unsigned int i;
1911
1912   if (key == NULL)
1913     {
1914 #if DEBUG_FS
1915       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1916                   "Received last result for `%s' from local datastore, deciding what to do next.\n",
1917                   GNUNET_h2s (&lgc->query));
1918 #endif
1919       /* no further results from datastore; continue
1920          processing further requests from the client and
1921          allow the next task to use the datastore; also,
1922          switch to P2P requests or clean up our state. */
1923       next_ds_request ();
1924       GNUNET_SERVER_receive_done (lgc->client,
1925                                   GNUNET_OK);
1926       if ( (lgc->results_used == 0) ||
1927            (lgc->type == GNUNET_DATASTORE_BLOCKTYPE_KBLOCK) ||
1928            (lgc->type == GNUNET_DATASTORE_BLOCKTYPE_SBLOCK) ||
1929            (lgc->type == GNUNET_DATASTORE_BLOCKTYPE_SKBLOCK) )
1930         {
1931 #if DEBUG_FS
1932           GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1933                       "Forwarding query for `%s' to network.\n",
1934                       GNUNET_h2s (&lgc->query));
1935 #endif
1936           cl = clients;
1937           while ( (NULL != cl) &&
1938                   (cl->client != lgc->client) )
1939             cl = cl->next;
1940           if (cl == NULL)
1941             {
1942               cl = GNUNET_malloc (sizeof (struct ClientList));
1943               cl->client = lgc->client;
1944               cl->next = clients;
1945               clients = cl;
1946             }
1947           crl = GNUNET_malloc (sizeof (struct ClientRequestList));
1948           crl->cl = cl;
1949           GNUNET_CONTAINER_DLL_insert (cl->head, cl->tail, crl);
1950           pr = GNUNET_malloc (sizeof (struct PendingRequest));
1951           pr->client = lgc->client;
1952           GNUNET_SERVER_client_keep (pr->client);
1953           pr->crl_entry = crl;
1954           crl->req = pr;
1955           if (lgc->type == GNUNET_DATASTORE_BLOCKTYPE_SBLOCK)
1956             {
1957               pr->namespace = GNUNET_malloc (sizeof (GNUNET_HashCode));
1958               *pr->namespace = lgc->namespace;
1959             }
1960           pr->replies_seen = lgc->results;
1961           lgc->results = NULL;
1962           pr->start_time = GNUNET_TIME_absolute_get ();
1963           pr->query = lgc->query;
1964           pr->target_pid = GNUNET_PEER_intern (&lgc->target);
1965           pr->replies_seen_off = lgc->results_used;
1966           pr->replies_seen_size = lgc->results_size;
1967           lgc->results_size = 0;
1968           pr->type = lgc->type;
1969           pr->anonymity_level = lgc->anonymity_level;
1970           pr->bf = refresh_bloomfilter (pr->replies_seen_off,
1971                                         &pr->mingle,
1972                                         &pr->bf_size,
1973                                         pr->replies_seen);
1974           GNUNET_CONTAINER_multihashmap_put (requests_by_query,
1975                                              &pr->query,
1976                                              pr,
1977                                              GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
1978           pr->task = GNUNET_SCHEDULER_add_delayed (sched,
1979                                                    get_processing_delay (),
1980                                                    &forward_request_task,
1981                                                    pr);
1982           local_get_context_free (lgc);
1983           return;
1984         }
1985       /* got all possible results, clean up! */
1986 #if DEBUG_FS
1987       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1988                   "Found all possible results for query for `%s', done!\n",
1989                   GNUNET_h2s (&lgc->query));
1990 #endif
1991       local_get_context_free (lgc);
1992       return;
1993     }
1994   if (type == GNUNET_DATASTORE_BLOCKTYPE_ONDEMAND)
1995     {
1996 #if DEBUG_FS
1997       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1998                   "Received on-demand block for `%s' from local datastore, fetching data.\n",
1999                   GNUNET_h2s (&lgc->query));
2000 #endif
2001       handle_on_demand_block (key, size, data, type, priority,
2002                               anonymity, expiration, uid,
2003                               &process_local_get_result,
2004                               lgc);
2005       return;
2006     }
2007   if ( (type != lgc->type) &&
2008        (lgc->type != GNUNET_DATASTORE_BLOCKTYPE_ANY) )
2009     {
2010       /* this should be virtually impossible to reach (DBLOCK
2011          query hash being identical to KBLOCK/SBLOCK query hash);
2012          nevertheless, if it happens, the correct thing is to
2013          simply skip the result. */
2014 #if DEBUG_FS
2015       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2016                   "Received block of unexpected type (%u, want %u) for `%s' from local datastore, ignoring.\n",
2017                   type,
2018                   lgc->type,
2019                   GNUNET_h2s (&lgc->query));
2020 #endif
2021       GNUNET_DATASTORE_get_next (dsh, GNUNET_YES);
2022       return;
2023     }
2024   /* check if this is a result we've alredy
2025      received */
2026   for (i=0;i<lgc->results_used;i++)
2027     if (0 == memcmp (key,
2028                      &lgc->results[i],
2029                      sizeof (GNUNET_HashCode)))
2030       {
2031 #if DEBUG_FS
2032         GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2033                     "Received duplicate result for `%s' from local datastore, ignoring.\n",
2034                     GNUNET_h2s (&lgc->query));
2035 #endif
2036         GNUNET_DATASTORE_get_next (dsh, GNUNET_YES);
2037         return;
2038       }
2039   if (lgc->results_used == lgc->results_size)
2040     GNUNET_array_grow (lgc->results,
2041                        lgc->results_size,
2042                        lgc->results_size * 2 + 2);
2043   GNUNET_CRYPTO_hash (data,
2044                       size,
2045                       &lgc->results[lgc->results_used++]);
2046   msize = size + sizeof (struct ContentMessage);
2047   GNUNET_assert (msize < GNUNET_SERVER_MAX_MESSAGE_SIZE);
2048   lgc->result = GNUNET_malloc (msize);
2049   lgc->result->header.size = htons (msize);
2050   lgc->result->header.type = htons (GNUNET_MESSAGE_TYPE_FS_CONTENT);
2051   lgc->result->type = htonl (type);
2052   lgc->result->expiration = GNUNET_TIME_absolute_hton (expiration);
2053   memcpy (&lgc->result[1],
2054           data,
2055           size);
2056 #if DEBUG_FS
2057   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2058               "Received new result for `%s' from local datastore, passing to client.\n",
2059               GNUNET_h2s (&lgc->query));
2060 #endif
2061   GNUNET_SERVER_notify_transmit_ready (lgc->client,
2062                                        msize,
2063                                        GNUNET_TIME_UNIT_FOREVER_REL,
2064                                        &transmit_local_result,
2065                                        lgc);
2066 }
2067
2068
2069 /**
2070  * We're processing a search request from a local
2071  * client.  Now it is our turn to query the datastore.
2072  *
2073  * @param cls our closure (struct LocalGetContext)
2074  * @param tc unused
2075  */
2076 static void
2077 transmit_local_get (void *cls,
2078                     const struct GNUNET_SCHEDULER_TaskContext *tc)
2079 {
2080   struct LocalGetContext *lgc = cls;
2081   uint32_t type;
2082
2083   type = lgc->type;
2084   if (type == GNUNET_DATASTORE_BLOCKTYPE_DBLOCK)
2085     type = GNUNET_DATASTORE_BLOCKTYPE_ANY; /* to get on-demand as well */
2086   GNUNET_DATASTORE_get (dsh,
2087                         &lgc->query,
2088                         type,
2089                         &process_local_get_result,
2090                         lgc,
2091                         GNUNET_TIME_UNIT_FOREVER_REL);
2092 }
2093
2094
2095 /**
2096  * We're processing a search request from a local
2097  * client.  Now it is our turn to query the datastore.
2098  *
2099  * @param cls our closure (struct LocalGetContext)
2100  * @param ok did we succeed to queue for datastore access, should always be GNUNET_OK
2101  */
2102 static void
2103 transmit_local_get_ready (void *cls,
2104                           int ok)
2105 {
2106   struct LocalGetContext *lgc = cls;
2107
2108   GNUNET_assert (GNUNET_OK == ok);
2109   GNUNET_SCHEDULER_add_continuation (sched,
2110                                      &transmit_local_get,
2111                                      lgc,
2112                                      GNUNET_SCHEDULER_REASON_PREREQ_DONE);
2113 }
2114
2115
2116 /**
2117  * Handle START_SEARCH-message (search request from client).
2118  *
2119  * @param cls closure
2120  * @param client identification of the client
2121  * @param message the actual message
2122  */
2123 static void
2124 handle_start_search (void *cls,
2125                      struct GNUNET_SERVER_Client *client,
2126                      const struct GNUNET_MessageHeader *message)
2127 {
2128   const struct SearchMessage *sm;
2129   struct LocalGetContext *lgc;
2130   uint16_t msize;
2131   unsigned int sc;
2132
2133   msize = ntohs (message->size);
2134   if ( (msize < sizeof (struct SearchMessage)) ||
2135        (0 != (msize - sizeof (struct SearchMessage)) % sizeof (GNUNET_HashCode)) )
2136     {
2137       GNUNET_break (0);
2138       GNUNET_SERVER_receive_done (client,
2139                                   GNUNET_SYSERR);
2140       return;
2141     }
2142   sc = (msize - sizeof (struct SearchMessage)) / sizeof (GNUNET_HashCode);
2143   sm = (const struct SearchMessage*) message;
2144   GNUNET_SERVER_client_keep (client);
2145   lgc = GNUNET_malloc (sizeof (struct LocalGetContext));
2146   if  (sc > 0)
2147     {
2148       lgc->results_used = sc;
2149       GNUNET_array_grow (lgc->results,
2150                          lgc->results_size,
2151                          sc * 2);
2152       memcpy (lgc->results,
2153               &sm[1],
2154               sc * sizeof (GNUNET_HashCode));
2155     }
2156   lgc->client = client;
2157   lgc->type = ntohl (sm->type);
2158   lgc->anonymity_level = ntohl (sm->anonymity_level);
2159   switch (lgc->type)
2160     {
2161     case GNUNET_DATASTORE_BLOCKTYPE_DBLOCK:
2162     case GNUNET_DATASTORE_BLOCKTYPE_IBLOCK:
2163       lgc->target.hashPubKey = sm->target;
2164       break;
2165     case GNUNET_DATASTORE_BLOCKTYPE_SBLOCK:
2166       lgc->namespace = sm->target;
2167       break;
2168     default:
2169       break;
2170     }
2171   lgc->query = sm->query;
2172   GNUNET_CONTAINER_DLL_insert (lgc_head, lgc_tail, lgc);
2173   lgc->req = queue_ds_request (GNUNET_TIME_UNIT_FOREVER_REL,
2174                                &transmit_local_get_ready,
2175                                lgc);
2176 }
2177
2178
2179 /**
2180  * List of handlers for the messages understood by this
2181  * service.
2182  */
2183 static struct GNUNET_SERVER_MessageHandler handlers[] = {
2184   {&handle_index_start, NULL,
2185    GNUNET_MESSAGE_TYPE_FS_INDEX_START, 0},
2186   {&handle_index_list_get, NULL,
2187    GNUNET_MESSAGE_TYPE_FS_INDEX_LIST_GET, sizeof(struct GNUNET_MessageHeader) },
2188   {&handle_unindex, NULL, GNUNET_MESSAGE_TYPE_FS_UNINDEX,
2189    sizeof (struct UnindexMessage) },
2190   {&handle_start_search, NULL, GNUNET_MESSAGE_TYPE_FS_START_SEARCH,
2191    0 },
2192   {NULL, NULL, 0, 0}
2193 };
2194
2195
2196 /**
2197  * Clean up the memory used by the PendingRequest structure (except
2198  * for the client or peer list that the request may be part of).
2199  *
2200  * @param pr request to clean up
2201  */
2202 static void
2203 destroy_pending_request (struct PendingRequest *pr)
2204 {
2205   struct PendingReply *reply;
2206   struct ClientList *cl;
2207
2208   GNUNET_CONTAINER_multihashmap_remove (requests_by_query,
2209                                         &pr->query,
2210                                         pr);
2211   // FIXME: not sure how this can work (efficiently)
2212   // also, what does the return value mean?
2213   if (pr->client == NULL)
2214     {
2215       GNUNET_CONTAINER_heap_remove_node (requests_by_expiration,
2216                                          pr);
2217     }
2218   else
2219     {
2220       cl = pr->crl_entry->cl;
2221       GNUNET_CONTAINER_DLL_remove (cl->head,
2222                                    cl->tail,
2223                                    pr->crl_entry);
2224     }
2225   if (GNUNET_SCHEDULER_NO_TASK != pr->task)
2226     GNUNET_SCHEDULER_cancel (sched, pr->task);
2227   if (NULL != pr->cth)
2228     GNUNET_CORE_notify_transmit_ready_cancel (pr->cth);
2229   if (NULL != pr->bf)
2230     GNUNET_CONTAINER_bloomfilter_free (pr->bf);
2231   if (NULL != pr->th)
2232     GNUNET_CONNECTION_notify_transmit_ready_cancel (pr->th);
2233   while (NULL != (reply = pr->replies_pending))
2234     {
2235       pr->replies_pending = reply->next;
2236       GNUNET_free (reply);
2237     }
2238   GNUNET_PEER_change_rc (pr->source_pid, -1);
2239   GNUNET_PEER_change_rc (pr->target_pid, -1);
2240   GNUNET_PEER_decrement_rcs (pr->used_pids, pr->used_pids_off);
2241   GNUNET_free_non_null (pr->used_pids);
2242   GNUNET_free_non_null (pr->replies_seen);
2243   GNUNET_free_non_null (pr->namespace);
2244   GNUNET_free (pr);
2245 }
2246
2247
2248 /**
2249  * A client disconnected.  Remove all of its pending queries.
2250  *
2251  * @param cls closure, NULL
2252  * @param client identification of the client
2253  */
2254 static void
2255 handle_client_disconnect (void *cls,
2256                           struct GNUNET_SERVER_Client
2257                           * client)
2258 {
2259   struct LocalGetContext *lgc;
2260   struct ClientList *cpos;
2261   struct ClientList *cprev;
2262   struct ClientRequestList *rl;
2263
2264   lgc = lgc_head;
2265   while ( (NULL != lgc) &&
2266           (lgc->client != client) )
2267     lgc = lgc->next;
2268   if (lgc != NULL)
2269     local_get_context_free (lgc);
2270   cprev = NULL;
2271   cpos = clients;
2272   while ( (NULL != cpos) &&
2273           (clients->client != client) )
2274     {
2275       cprev = cpos;
2276       cpos = cpos->next;
2277     }
2278   if (cpos != NULL)
2279     {
2280       if (cprev == NULL)
2281         clients = cpos->next;
2282       else
2283         cprev->next = cpos->next;
2284       while (NULL != (rl = cpos->head))
2285         {
2286           cpos->head = rl->next;
2287           destroy_pending_request (rl->req);
2288           GNUNET_free (rl);
2289         }
2290       GNUNET_free (cpos);
2291     }
2292 }
2293
2294
2295 /**
2296  * Iterator over entries in the "requests_by_query" map
2297  * that frees all the entries.
2298  *
2299  * @param cls closure, NULL
2300  * @param key current key code (the query, unused)
2301  * @param value value in the hash map, of type "struct PendingRequest*"
2302  * @return GNUNET_YES (we should continue to  iterate)
2303  */
2304 static int
2305 destroy_pending_request_cb (void *cls,
2306                             const GNUNET_HashCode * key,
2307                             void *value)
2308 {
2309   struct PendingRequest *pr = value;
2310
2311   destroy_pending_request (pr);
2312   return GNUNET_YES;
2313 }
2314
2315
2316 /**
2317  * Task run during shutdown.
2318  *
2319  * @param cls unused
2320  * @param tc unused
2321  */
2322 static void
2323 shutdown_task (void *cls,
2324                const struct GNUNET_SCHEDULER_TaskContext *tc)
2325 {
2326   struct IndexInfo *pos;
2327
2328   if (NULL != core)
2329     GNUNET_CORE_disconnect (core);
2330   GNUNET_DATASTORE_disconnect (dsh,
2331                                GNUNET_NO);
2332   dsh = NULL;
2333   GNUNET_CONTAINER_multihashmap_iterate (requests_by_query,
2334                                          &destroy_pending_request_cb,
2335                                          NULL);
2336   while (clients != NULL)
2337     handle_client_disconnect (NULL,
2338                               clients->client);
2339   GNUNET_CONTAINER_multihashmap_destroy (requests_by_query);
2340   requests_by_query = NULL;
2341   GNUNET_CONTAINER_multihashmap_destroy (requests_by_peer);
2342   requests_by_peer = NULL;
2343   GNUNET_CONTAINER_heap_destroy (requests_by_expiration);
2344   requests_by_expiration = NULL;
2345   // FIXME: iterate over entries and free individually?
2346   // (or do we get disconnect notifications?)
2347   GNUNET_CONTAINER_multihashmap_destroy (connected_peers);
2348   connected_peers = NULL;
2349   GNUNET_CONTAINER_multihashmap_destroy (ifm);
2350   ifm = NULL;
2351   while (NULL != (pos = indexed_files))
2352     {
2353       indexed_files = pos->next;
2354       GNUNET_free (pos);
2355     }
2356 }
2357
2358
2359 /**
2360  * Free (each) request made by the peer.
2361  *
2362  * @param cls closure, points to peer that the request belongs to
2363  * @param key current key code
2364  * @param value value in the hash map
2365  * @return GNUNET_YES (we should continue to iterate)
2366  */
2367 static int
2368 destroy_request (void *cls,
2369                  const GNUNET_HashCode * key,
2370                  void *value)
2371 {
2372   const struct GNUNET_PeerIdentity * peer = cls;
2373   struct PendingRequest *pr = value;
2374
2375   GNUNET_CONTAINER_multihashmap_remove (requests_by_peer,
2376                                         &peer->hashPubKey,
2377                                         pr);
2378   destroy_pending_request (pr);
2379   return GNUNET_YES;
2380 }
2381
2382
2383
2384 /**
2385  * Method called whenever a given peer connects.
2386  *
2387  * @param cls closure, not used
2388  * @param peer peer identity this notification is about
2389  */
2390 static void
2391 peer_connect_handler (void *cls,
2392                       const struct
2393                       GNUNET_PeerIdentity * peer)
2394 {
2395   struct ConnectedPeer *cp;
2396
2397   cp = GNUNET_malloc (sizeof (struct ConnectedPeer));
2398   cp->pid = GNUNET_PEER_intern (peer);
2399   GNUNET_CONTAINER_multihashmap_put (connected_peers,
2400                                      &peer->hashPubKey,
2401                                      cp,
2402                                      GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY);
2403 }
2404
2405
2406 /**
2407  * Method called whenever a peer disconnects.
2408  *
2409  * @param cls closure, not used
2410  * @param peer peer identity this notification is about
2411  */
2412 static void
2413 peer_disconnect_handler (void *cls,
2414                          const struct
2415                          GNUNET_PeerIdentity * peer)
2416 {
2417   struct ConnectedPeer *cp;
2418
2419   cp = GNUNET_CONTAINER_multihashmap_get (connected_peers,
2420                                           &peer->hashPubKey);
2421   GNUNET_PEER_change_rc (cp->pid, -1);
2422   GNUNET_PEER_decrement_rcs (cp->last_p2p_replies, P2P_SUCCESS_LIST_SIZE);
2423   GNUNET_free (cp);
2424   GNUNET_CONTAINER_multihashmap_get_multiple (requests_by_peer,
2425                                               &peer->hashPubKey,
2426                                               &destroy_request,
2427                                               (void*) peer);
2428 }
2429
2430
2431 /**
2432  * We're processing a GET request from
2433  * another peer and have decided to forward
2434  * it to other peers.
2435  *
2436  * @param cls our "struct ProcessGetContext *"
2437  * @param tc unused
2438  */
2439 static void
2440 forward_get_request (void *cls,
2441                      const struct GNUNET_SCHEDULER_TaskContext *tc)
2442 {
2443   struct ProcessGetContext *pgc = cls;
2444   struct PendingRequest *pr;
2445   struct PendingRequest *eer;
2446   struct GNUNET_PeerIdentity target;
2447
2448   pr = GNUNET_malloc (sizeof (struct PendingRequest));
2449   if (GET_MESSAGE_BIT_SKS_NAMESPACE == (GET_MESSAGE_BIT_SKS_NAMESPACE & pgc->bm))
2450     {
2451       pr->namespace = GNUNET_malloc (sizeof(GNUNET_HashCode));
2452       *pr->namespace = pgc->namespace;
2453     }
2454   pr->bf = pgc->bf;
2455   pr->bf_size = pgc->bf_size;
2456   pgc->bf = NULL;
2457   pr->start_time = pgc->start_time;
2458   pr->query = pgc->query;
2459   pr->source_pid = GNUNET_PEER_intern (&pgc->reply_to);
2460   if (GET_MESSAGE_BIT_TRANSMIT_TO == (GET_MESSAGE_BIT_TRANSMIT_TO & pgc->bm))
2461     pr->target_pid = GNUNET_PEER_intern (&pgc->prime_target);
2462   pr->anonymity_level = 1; /* default */
2463   pr->priority = pgc->priority;
2464   pr->remaining_priority = pr->priority;
2465   pr->mingle = pgc->mingle;
2466   pr->ttl = pgc->ttl;
2467   pr->type = pgc->type;
2468   GNUNET_CONTAINER_multihashmap_put (requests_by_query,
2469                                      &pr->query,
2470                                      pr,
2471                                      GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
2472   GNUNET_CONTAINER_multihashmap_put (requests_by_peer,
2473                                      &pgc->reply_to.hashPubKey,
2474                                      pr,
2475                                      GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
2476   GNUNET_CONTAINER_heap_insert (requests_by_expiration,
2477                                 pr,
2478                                 pr->start_time.value + pr->ttl);
2479   if (GNUNET_CONTAINER_heap_get_size (requests_by_expiration) > max_pending_requests)
2480     {
2481       /* expire oldest request! */
2482       eer = GNUNET_CONTAINER_heap_peek (requests_by_expiration);
2483       GNUNET_PEER_resolve (eer->source_pid,
2484                            &target);
2485       GNUNET_CONTAINER_multihashmap_remove (requests_by_peer,
2486                                             &target.hashPubKey,
2487                                             eer);
2488       destroy_pending_request (eer);
2489     }
2490   pr->task = GNUNET_SCHEDULER_add_delayed (sched,
2491                                            get_processing_delay (),
2492                                            &forward_request_task,
2493                                            pr);
2494   GNUNET_free (pgc);
2495 }
2496 /**
2497  * Transmit the given message by copying it to
2498  * the target buffer "buf".  "buf" will be
2499  * NULL and "size" zero if the socket was closed for
2500  * writing in the meantime.  In that case, only
2501
2502  * free the message
2503  *
2504  * @param cls closure, pointer to the message
2505  * @param size number of bytes available in buf
2506  * @param buf where the callee should write the message
2507  * @return number of bytes written to buf
2508  */
2509 static size_t
2510 transmit_message (void *cls,
2511                   size_t size, void *buf)
2512 {
2513   struct GNUNET_MessageHeader *msg = cls;
2514   uint16_t msize;
2515
2516   if (NULL == buf)
2517     {
2518 #if DEBUG_FS
2519       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2520                   "Dropping reply, core too busy.\n");
2521 #endif
2522       GNUNET_free (msg);
2523       return 0;
2524     }
2525   msize = ntohs (msg->size);
2526   GNUNET_assert (size >= msize);
2527   memcpy (buf, msg, msize);
2528   GNUNET_free (msg);
2529   return msize;
2530 }
2531
2532
2533 /**
2534  * Test if the load on this peer is too high
2535  * to even consider processing the query at
2536  * all.
2537  *
2538  * @return GNUNET_YES if the load is too high, GNUNET_NO otherwise
2539  */
2540 static int
2541 test_load_too_high ()
2542 {
2543   return GNUNET_NO; // FIXME
2544 }
2545
2546
2547 /**
2548  * We're processing (local) results for a search request
2549  * from another peer.  Pass applicable results to the
2550  * peer and if we are done either clean up (operation
2551  * complete) or forward to other peers (more results possible).
2552  *
2553  * @param cls our closure (struct LocalGetContext)
2554  * @param key key for the content
2555  * @param size number of bytes in data
2556  * @param data content stored
2557  * @param type type of the content
2558  * @param priority priority of the content
2559  * @param anonymity anonymity-level for the content
2560  * @param expiration expiration time for the content
2561  * @param uid unique identifier for the datum;
2562  *        maybe 0 if no unique identifier is available
2563  */
2564 static void
2565 process_p2p_get_result (void *cls,
2566                         const GNUNET_HashCode * key,
2567                         uint32_t size,
2568                         const void *data,
2569                         uint32_t type,
2570                         uint32_t priority,
2571                         uint32_t anonymity,
2572                         struct GNUNET_TIME_Absolute
2573                         expiration,
2574                         uint64_t uid)
2575 {
2576   struct ProcessGetContext *pgc = cls;
2577   GNUNET_HashCode dhash;
2578   GNUNET_HashCode mhash;
2579   struct PutMessage *reply;
2580
2581   if (NULL == key)
2582     {
2583       /* no more results */
2584       if ( ( (pgc->policy & ROUTING_POLICY_FORWARD) ==  ROUTING_POLICY_FORWARD) &&
2585            ( (0 == pgc->results_found) ||
2586              (pgc->type == GNUNET_DATASTORE_BLOCKTYPE_KBLOCK) ||
2587              (pgc->type == GNUNET_DATASTORE_BLOCKTYPE_SBLOCK) ||
2588              (pgc->type == GNUNET_DATASTORE_BLOCKTYPE_SKBLOCK) ) )
2589         {
2590           GNUNET_SCHEDULER_add_continuation (sched,
2591                                              &forward_get_request,
2592                                              pgc,
2593                                              GNUNET_SCHEDULER_REASON_PREREQ_DONE);
2594         }
2595       else
2596         {
2597           if (pgc->bf != NULL)
2598             GNUNET_CONTAINER_bloomfilter_free (pgc->bf);
2599           GNUNET_free (pgc);
2600         }
2601       next_ds_request ();
2602       return;
2603     }
2604   if (type == GNUNET_DATASTORE_BLOCKTYPE_ONDEMAND)
2605     {
2606       handle_on_demand_block (key, size, data, type, priority,
2607                               anonymity, expiration, uid,
2608                               &process_p2p_get_result,
2609                               pgc);
2610       return;
2611     }
2612   /* check for duplicates */
2613   GNUNET_CRYPTO_hash (data, size, &dhash);
2614   mingle_hash (&dhash,
2615                pgc->mingle,
2616                &mhash);
2617   if ( (pgc->bf != NULL) &&
2618        (GNUNET_YES ==
2619         GNUNET_CONTAINER_bloomfilter_test (pgc->bf,
2620                                            &mhash)) )
2621     {
2622 #if DEBUG_FS
2623       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2624                   "Result from datastore filtered by bloomfilter.\n");
2625 #endif
2626       GNUNET_DATASTORE_get_next (dsh, GNUNET_YES);
2627       return;
2628     }
2629   pgc->results_found++;
2630   if ( (pgc->type == GNUNET_DATASTORE_BLOCKTYPE_KBLOCK) ||
2631        (pgc->type == GNUNET_DATASTORE_BLOCKTYPE_SBLOCK) ||
2632        (pgc->type == GNUNET_DATASTORE_BLOCKTYPE_SKBLOCK) )
2633     {
2634       if (pgc->bf == NULL)
2635         {
2636           pgc->bf_size = 32;
2637           pgc->bf = GNUNET_CONTAINER_bloomfilter_init (NULL,
2638                                                        pgc->bf_size,
2639                                                        BLOOMFILTER_K);
2640         }
2641       GNUNET_CONTAINER_bloomfilter_add (pgc->bf,
2642                                         &mhash);
2643     }
2644
2645   reply = GNUNET_malloc (sizeof (struct PutMessage) + size);
2646   reply->header.size = htons (sizeof (struct PutMessage) + size);
2647   reply->header.type = htons (GNUNET_MESSAGE_TYPE_FS_PUT);
2648   reply->type = htonl (type);
2649   reply->expiration = GNUNET_TIME_relative_hton (GNUNET_TIME_absolute_get_remaining (expiration));
2650   memcpy (&reply[1], data, size);
2651   GNUNET_CORE_notify_transmit_ready (core,
2652                                      pgc->priority,
2653                                      ACCEPTABLE_REPLY_DELAY,
2654                                      &pgc->reply_to,
2655                                      sizeof (struct PutMessage) + size,
2656                                      &transmit_message,
2657                                      reply);
2658   if ( (GNUNET_YES == test_load_too_high()) ||
2659        (pgc->results_found > 5 + 2 * pgc->priority) )
2660     {
2661       GNUNET_DATASTORE_get_next (dsh, GNUNET_NO);
2662       pgc->policy &= ~ ROUTING_POLICY_FORWARD;
2663       return;
2664     }
2665   GNUNET_DATASTORE_get_next (dsh, GNUNET_YES);
2666 }
2667
2668
2669 /**
2670  * We're processing a GET request from another peer.  Give it to our
2671  * local datastore.
2672  *
2673  * @param cls our "struct ProcessGetContext"
2674  * @param ok did we get a datastore slice or not?
2675  */
2676 static void
2677 ds_get_request (void *cls,
2678                 int ok)
2679 {
2680   struct ProcessGetContext *pgc = cls;
2681   uint32_t type;
2682   struct GNUNET_TIME_Relative timeout;
2683
2684   if (GNUNET_OK != ok)
2685     {
2686       /* no point in doing P2P stuff if we can't even do local */
2687       GNUNET_free (dsh);
2688       return;
2689     }
2690   type = pgc->type;
2691   if (type == GNUNET_DATASTORE_BLOCKTYPE_DBLOCK)
2692     type = GNUNET_DATASTORE_BLOCKTYPE_ANY; /* to get on-demand as well */
2693   timeout = GNUNET_TIME_relative_multiply (BASIC_DATASTORE_REQUEST_DELAY,
2694                                            (pgc->priority + 1));
2695   GNUNET_DATASTORE_get (dsh,
2696                         &pgc->query,
2697                         type,
2698                         &process_p2p_get_result,
2699                         pgc,
2700                         timeout);
2701 }
2702
2703
2704 /**
2705  * The priority level imposes a bound on the maximum
2706  * value for the ttl that can be requested.
2707  *
2708  * @param ttl_in requested ttl
2709  * @param prio given priority
2710  * @return ttl_in if ttl_in is below the limit,
2711  *         otherwise the ttl-limit for the given priority
2712  */
2713 static int32_t
2714 bound_ttl (int32_t ttl_in, uint32_t prio)
2715 {
2716   unsigned long long allowed;
2717
2718   if (ttl_in <= 0)
2719     return ttl_in;
2720   allowed = ((unsigned long long) prio) * TTL_DECREMENT / 1000;
2721   if (ttl_in > allowed)
2722     {
2723       if (allowed >= (1 << 30))
2724         return 1 << 30;
2725       return allowed;
2726     }
2727   return ttl_in;
2728 }
2729
2730
2731 /**
2732  * We've received a request with the specified
2733  * priority.  Bound it according to how much
2734  * we trust the given peer.
2735  *
2736  * @param prio_in requested priority
2737  * @param peer the peer making the request
2738  * @return effective priority
2739  */
2740 static uint32_t
2741 bound_priority (uint32_t prio_in,
2742                 const struct GNUNET_PeerIdentity *peer)
2743 {
2744   return 0; // FIXME!
2745 }
2746
2747
2748 /**
2749  * Handle P2P "GET" request.
2750  *
2751  * @param cls closure, always NULL
2752  * @param other the other peer involved (sender or receiver, NULL
2753  *        for loopback messages where we are both sender and receiver)
2754  * @param message the actual message
2755  * @return GNUNET_OK to keep the connection open,
2756  *         GNUNET_SYSERR to close it (signal serious error)
2757  */
2758 static int
2759 handle_p2p_get (void *cls,
2760                 const struct GNUNET_PeerIdentity *other,
2761                 const struct GNUNET_MessageHeader *message)
2762 {
2763   uint16_t msize;
2764   const struct GetMessage *gm;
2765   unsigned int bits;
2766   const GNUNET_HashCode *opt;
2767   struct ProcessGetContext *pgc;
2768   uint32_t bm;
2769   size_t bfsize;
2770   uint32_t ttl_decrement;
2771   double preference;
2772   int net_load_up;
2773   int net_load_down;
2774
2775   msize = ntohs(message->size);
2776   if (msize < sizeof (struct GetMessage))
2777     {
2778       GNUNET_break_op (0);
2779       return GNUNET_SYSERR;
2780     }
2781   gm = (const struct GetMessage*) message;
2782   bm = ntohl (gm->hash_bitmap);
2783   bits = 0;
2784   while (bm > 0)
2785     {
2786       if (1 == (bm & 1))
2787         bits++;
2788       bm >>= 1;
2789     }
2790   if (msize < sizeof (struct GetMessage) + bits * sizeof (GNUNET_HashCode))
2791     {
2792       GNUNET_break_op (0);
2793       return GNUNET_SYSERR;
2794     }
2795   opt = (const GNUNET_HashCode*) &gm[1];
2796   bfsize = msize - sizeof (struct GetMessage) + bits * sizeof (GNUNET_HashCode);
2797   pgc = GNUNET_malloc (sizeof (struct ProcessGetContext));
2798   if (bfsize > 0)
2799     {
2800       pgc->bf = GNUNET_CONTAINER_bloomfilter_init ((const char*) &pgc[1],
2801                                                    bfsize,
2802                                                    BLOOMFILTER_K);
2803       pgc->bf_size = bfsize;
2804     }
2805   pgc->type = ntohl (gm->type);
2806   pgc->bm = ntohl (gm->hash_bitmap);
2807   pgc->mingle = gm->filter_mutator;
2808   bits = 0;
2809   if (0 != (pgc->bm & GET_MESSAGE_BIT_RETURN_TO))
2810     pgc->reply_to.hashPubKey = opt[bits++];
2811   else
2812     pgc->reply_to = *other;
2813   if (0 != (pgc->bm & GET_MESSAGE_BIT_SKS_NAMESPACE))
2814     pgc->namespace = opt[bits++];
2815   else if (pgc->type == GNUNET_DATASTORE_BLOCKTYPE_SBLOCK)
2816     {
2817       GNUNET_break_op (0);
2818       GNUNET_free (pgc);
2819       return GNUNET_SYSERR;
2820     }
2821   if (0 != (pgc->bm & GET_MESSAGE_BIT_TRANSMIT_TO))
2822     pgc->prime_target.hashPubKey = opt[bits++];
2823   /* note that we can really only check load here since otherwise
2824      peers could find out that we are overloaded by being disconnected
2825      after sending us a malformed query... */
2826   if (GNUNET_YES == test_load_too_high ())
2827     {
2828       if (NULL != pgc->bf)
2829         GNUNET_CONTAINER_bloomfilter_free (pgc->bf);
2830       GNUNET_free (pgc);
2831 #if DEBUG_FS
2832       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2833                   "Dropping query from `%s', this peer is too busy.\n",
2834                   GNUNET_i2s (other));
2835 #endif
2836       return GNUNET_OK;
2837     }
2838   net_load_up = 50; // FIXME
2839   net_load_down = 50; // FIXME
2840   pgc->policy = ROUTING_POLICY_NONE;
2841   if ( (net_load_up < IDLE_LOAD_THRESHOLD) &&
2842        (net_load_down < IDLE_LOAD_THRESHOLD) )
2843     {
2844       pgc->policy |= ROUTING_POLICY_ALL;
2845       pgc->priority = 0; /* no charge */
2846     }
2847   else
2848     {
2849       pgc->priority = bound_priority (ntohl (gm->priority), other);
2850       if ( (net_load_up <
2851             IDLE_LOAD_THRESHOLD + pgc->priority * pgc->priority) &&
2852            (net_load_down <
2853             IDLE_LOAD_THRESHOLD + pgc->priority * pgc->priority) )
2854         {
2855           pgc->policy |= ROUTING_POLICY_ALL;
2856         }
2857       else
2858         {
2859           // FIXME: is this sound?
2860           if (net_load_up < 90 + 10 * pgc->priority)
2861             pgc->policy |= ROUTING_POLICY_FORWARD;
2862           if (net_load_down < 90 + 10 * pgc->priority)
2863             pgc->policy |= ROUTING_POLICY_ANSWER;
2864         }
2865     }
2866   if (pgc->policy == ROUTING_POLICY_NONE)
2867     {
2868 #if DEBUG_FS
2869       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2870                   "Dropping query from `%s', network saturated.\n",
2871                   GNUNET_i2s (other));
2872 #endif
2873       if (NULL != pgc->bf)
2874         GNUNET_CONTAINER_bloomfilter_free (pgc->bf);
2875       GNUNET_free (pgc);
2876       return GNUNET_OK;     /* drop */
2877     }
2878   if ((pgc->policy & ROUTING_POLICY_INDIRECT) != ROUTING_POLICY_INDIRECT)
2879     pgc->priority = 0;  /* kill the priority (we cannot benefit) */
2880   pgc->ttl = bound_ttl (ntohl (gm->ttl), pgc->priority);
2881   /* decrement ttl (always) */
2882   ttl_decrement = 2 * TTL_DECREMENT +
2883     GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
2884                               TTL_DECREMENT);
2885   if ( (pgc->ttl < 0) &&
2886        (pgc->ttl - ttl_decrement > 0) )
2887     {
2888 #if DEBUG_FS
2889       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2890                   "Dropping query from `%s' due to TTL underflow.\n",
2891                   GNUNET_i2s (other));
2892 #endif
2893       /* integer underflow => drop (should be very rare)! */
2894       if (NULL != pgc->bf)
2895         GNUNET_CONTAINER_bloomfilter_free (pgc->bf);
2896       GNUNET_free (pgc);
2897       return GNUNET_OK;
2898     }
2899   pgc->ttl -= ttl_decrement;
2900   pgc->start_time = GNUNET_TIME_absolute_get ();
2901   preference = (double) pgc->priority;
2902   if (preference < QUERY_BANDWIDTH_VALUE)
2903     preference = QUERY_BANDWIDTH_VALUE;
2904   // FIXME: also reserve bandwidth for reply?
2905   GNUNET_CORE_peer_configure (core,
2906                               other,
2907                               GNUNET_TIME_UNIT_FOREVER_REL,
2908                               0, 0, preference, NULL, NULL);
2909   if (0 != (pgc->policy & ROUTING_POLICY_ANSWER))
2910     pgc->drq = queue_ds_request (BASIC_DATASTORE_REQUEST_DELAY,
2911                                  &ds_get_request,
2912                                  pgc);
2913   else
2914     GNUNET_SCHEDULER_add_continuation (sched,
2915                                        &forward_get_request,
2916                                        pgc,
2917                                        GNUNET_SCHEDULER_REASON_PREREQ_DONE);
2918   return GNUNET_OK;
2919 }
2920
2921
2922 /**
2923  * Function called to notify us that we can now transmit a reply to a
2924  * client or peer.  "buf" will be NULL and "size" zero if the socket was
2925  * closed for writing in the meantime.
2926  *
2927  * @param cls closure, points to a "struct PendingRequest*" with
2928  *            one or more pending replies
2929  * @param size number of bytes available in buf
2930  * @param buf where the callee should write the message
2931  * @return number of bytes written to buf
2932  */
2933 static size_t
2934 transmit_result (void *cls,
2935                  size_t size,
2936                  void *buf)
2937 {
2938   struct PendingRequest *pr = cls;
2939   char *cbuf = buf;
2940   struct PendingReply *reply;
2941   size_t ret;
2942
2943   ret = 0;
2944   while (NULL != (reply = pr->replies_pending))
2945     {
2946       if ( (reply->msize + ret < ret) ||
2947            (reply->msize + ret > size) )
2948         break;
2949       pr->replies_pending = reply->next;
2950       memcpy (&cbuf[ret], &reply[1], reply->msize);
2951       ret += reply->msize;
2952       GNUNET_free (reply);
2953     }
2954   return ret;
2955 }
2956
2957
2958 /**
2959  * Iterator over pending requests.
2960  *
2961  * @param cls response (struct ProcessReplyClosure)
2962  * @param key our query
2963  * @param value value in the hash map (meta-info about the query)
2964  * @return GNUNET_YES (we should continue to iterate)
2965  */
2966 static int
2967 process_reply (void *cls,
2968                const GNUNET_HashCode * key,
2969                void *value)
2970 {
2971   struct ProcessReplyClosure *prq = cls;
2972   struct PendingRequest *pr = value;
2973   struct PendingRequest *eer;
2974   struct PendingReply *reply;
2975   struct PutMessage *pm;
2976   struct ContentMessage *cm;
2977   GNUNET_HashCode chash;
2978   GNUNET_HashCode mhash;
2979   struct GNUNET_PeerIdentity target;
2980   size_t msize;
2981   uint32_t prio;
2982   struct GNUNET_TIME_Relative max_delay;
2983
2984   GNUNET_CRYPTO_hash (prq->data,
2985                       prq->size,
2986                       &chash);
2987   switch (prq->type)
2988     {
2989     case GNUNET_DATASTORE_BLOCKTYPE_DBLOCK:
2990     case GNUNET_DATASTORE_BLOCKTYPE_IBLOCK:
2991       break;
2992     case GNUNET_DATASTORE_BLOCKTYPE_SBLOCK:
2993       /* FIXME: does prq->namespace match our expectations? */
2994       /* then: fall-through??? */
2995     case GNUNET_DATASTORE_BLOCKTYPE_KBLOCK:
2996       if (pr->bf != NULL)
2997         {
2998           mingle_hash (&chash, pr->mingle, &mhash);
2999           if (GNUNET_YES == GNUNET_CONTAINER_bloomfilter_test (pr->bf,
3000                                                                &mhash))
3001             return GNUNET_YES; /* duplicate */
3002           GNUNET_CONTAINER_bloomfilter_add (pr->bf,
3003                                             &mhash);
3004         }
3005       break;
3006     case GNUNET_DATASTORE_BLOCKTYPE_SKBLOCK:
3007       // FIXME: any checks against duplicates for SKBlocks?
3008       break;
3009     }
3010   prio = pr->priority;
3011   prq->priority += pr->remaining_priority;
3012   pr->remaining_priority = 0;
3013   if (pr->client != NULL)
3014     {
3015       if (pr->replies_seen_size == pr->replies_seen_off)
3016         GNUNET_array_grow (pr->replies_seen,
3017                            pr->replies_seen_size,
3018                            pr->replies_seen_size * 2 + 4);
3019       pr->replies_seen[pr->replies_seen_off++] = chash;
3020       // FIXME: possibly recalculate BF!
3021     }
3022   if (pr->client == NULL)
3023     {
3024       msize = sizeof (struct ContentMessage) + prq->size;
3025       reply = GNUNET_malloc (msize + sizeof (struct PendingReply));
3026       reply->msize = msize;
3027       cm = (struct ContentMessage*) &reply[1];
3028       cm->header.type = htons (GNUNET_MESSAGE_TYPE_FS_CONTENT);
3029       cm->header.size = htons (msize);
3030       cm->type = htonl (prq->type);
3031       cm->expiration = GNUNET_TIME_absolute_hton (prq->expiration);
3032       reply->next = pr->replies_pending;
3033       pr->replies_pending = reply;
3034       memcpy (&reply[1], prq->data, prq->size);
3035       if (pr->cth != NULL)
3036         return GNUNET_YES;
3037       max_delay = GNUNET_TIME_UNIT_FOREVER_REL;
3038       if (GNUNET_CONTAINER_heap_get_size (requests_by_expiration) >= max_pending_requests)
3039         {
3040           /* estimate expiration time from time difference between
3041              first request that will be discarded and this request */
3042           eer = GNUNET_CONTAINER_heap_peek (requests_by_expiration);
3043           max_delay = GNUNET_TIME_absolute_get_difference (pr->start_time,
3044                                                            eer->start_time);
3045         }
3046       GNUNET_PEER_resolve (pr->source_pid,
3047                            &target);
3048       pr->cth = GNUNET_CORE_notify_transmit_ready (core,
3049                                                    prio,
3050                                                    max_delay,
3051                                                    &target,
3052                                                    msize,
3053                                                    &transmit_result,
3054                                                    pr);
3055       if (NULL == pr->cth)
3056         {
3057           // FIXME: now what? discard?
3058         }
3059     }
3060   else
3061     {
3062       msize = sizeof (struct PutMessage) + prq->size;
3063       reply = GNUNET_malloc (msize + sizeof (struct PendingReply));
3064       reply->msize = msize;
3065       reply->next = pr->replies_pending;
3066       pm = (struct PutMessage*) &reply[1];
3067       pm->header.type = htons (GNUNET_MESSAGE_TYPE_FS_PUT);
3068       pm->header.size = htons (msize);
3069       pm->type = htonl (prq->type);
3070       pm->expiration = GNUNET_TIME_relative_hton (GNUNET_TIME_absolute_get_remaining (prq->expiration));
3071       pr->replies_pending = reply;
3072       memcpy (&reply[1], prq->data, prq->size);
3073       if (pr->th != NULL)
3074         return GNUNET_YES;
3075       pr->th = GNUNET_SERVER_notify_transmit_ready (pr->client,
3076                                                     msize,
3077                                                     GNUNET_TIME_UNIT_FOREVER_REL,
3078                                                     &transmit_result,
3079                                                     pr);
3080       if (pr->th == NULL)
3081         {
3082           // FIXME: need to try again later (not much
3083           // to do here specifically, but we need to
3084           // check somewhere else to handle this case!)
3085         }
3086     }
3087   // FIXME: implement hot-path routing statistics keeping!
3088   return GNUNET_YES;
3089 }
3090
3091
3092 /**
3093  * Check if the given KBlock is well-formed.
3094  *
3095  * @param kb the kblock data (or at least "dsize" bytes claiming to be one)
3096  * @param dsize size of "kb" in bytes; check for < sizeof(struct KBlock)!
3097  * @param query where to store the query that this block answers
3098  * @return GNUNET_OK if this is actually a well-formed KBlock
3099  */
3100 static int
3101 check_kblock (const struct KBlock *kb,
3102               size_t dsize,
3103               GNUNET_HashCode *query)
3104 {
3105   if (dsize < sizeof (struct KBlock))
3106     {
3107       GNUNET_break_op (0);
3108       return GNUNET_SYSERR;
3109     }
3110   if (dsize - sizeof (struct KBlock) !=
3111       ntohs (kb->purpose.size)
3112       - sizeof (struct GNUNET_CRYPTO_RsaSignaturePurpose)
3113       - sizeof(struct GNUNET_CRYPTO_RsaPublicKeyBinaryEncoded) )
3114     {
3115       GNUNET_break_op (0);
3116       return GNUNET_SYSERR;
3117     }
3118   if (GNUNET_OK !=
3119       GNUNET_CRYPTO_rsa_verify (GNUNET_SIGNATURE_PURPOSE_FS_KBLOCK,
3120                                 &kb->purpose,
3121                                 &kb->signature,
3122                                 &kb->keyspace))
3123     {
3124       GNUNET_break_op (0);
3125       return GNUNET_SYSERR;
3126     }
3127   if (query != NULL)
3128     GNUNET_CRYPTO_hash (&kb->keyspace,
3129                         sizeof (struct GNUNET_CRYPTO_RsaPublicKeyBinaryEncoded),
3130                         query);
3131   return GNUNET_OK;
3132 }
3133
3134
3135 /**
3136  * Check if the given SBlock is well-formed.
3137  *
3138  * @param sb the sblock data (or at least "dsize" bytes claiming to be one)
3139  * @param dsize size of "kb" in bytes; check for < sizeof(struct SBlock)!
3140  * @param query where to store the query that this block answers
3141  * @param namespace where to store the namespace that this block belongs to
3142  * @return GNUNET_OK if this is actually a well-formed SBlock
3143  */
3144 static int
3145 check_sblock (const struct SBlock *sb,
3146               size_t dsize,
3147               GNUNET_HashCode *query,
3148               GNUNET_HashCode *namespace)
3149 {
3150   if (dsize < sizeof (struct SBlock))
3151     {
3152       GNUNET_break_op (0);
3153       return GNUNET_SYSERR;
3154     }
3155   if (dsize !=
3156       ntohs (sb->purpose.size) + sizeof (struct GNUNET_CRYPTO_RsaSignature))
3157     {
3158       GNUNET_break_op (0);
3159       return GNUNET_SYSERR;
3160     }
3161   if (GNUNET_OK !=
3162       GNUNET_CRYPTO_rsa_verify (GNUNET_SIGNATURE_PURPOSE_FS_SBLOCK,
3163                                 &sb->purpose,
3164                                 &sb->signature,
3165                                 &sb->subspace))
3166     {
3167       GNUNET_break_op (0);
3168       return GNUNET_SYSERR;
3169     }
3170   if (query != NULL)
3171     *query = sb->identifier;
3172   if (namespace != NULL)
3173     GNUNET_CRYPTO_hash (&sb->subspace,
3174                         sizeof (struct GNUNET_CRYPTO_RsaPublicKeyBinaryEncoded),
3175                         namespace);
3176   return GNUNET_OK;
3177 }
3178
3179
3180 /**
3181  * Handle P2P "PUT" request.
3182  *
3183  * @param cls closure, always NULL
3184  * @param other the other peer involved (sender or receiver, NULL
3185  *        for loopback messages where we are both sender and receiver)
3186  * @param message the actual message
3187  * @return GNUNET_OK to keep the connection open,
3188  *         GNUNET_SYSERR to close it (signal serious error)
3189  */
3190 static int
3191 handle_p2p_put (void *cls,
3192                 const struct GNUNET_PeerIdentity *other,
3193                 const struct GNUNET_MessageHeader *message)
3194 {
3195   const struct PutMessage *put;
3196   uint16_t msize;
3197   size_t dsize;
3198   uint32_t type;
3199   struct GNUNET_TIME_Absolute expiration;
3200   GNUNET_HashCode query;
3201   struct ProcessReplyClosure prq;
3202
3203   msize = ntohs (message->size);
3204   if (msize < sizeof (struct PutMessage))
3205     {
3206       GNUNET_break_op(0);
3207       return GNUNET_SYSERR;
3208     }
3209   put = (const struct PutMessage*) message;
3210   dsize = msize - sizeof (struct PutMessage);
3211   type = ntohl (put->type);
3212   expiration = GNUNET_TIME_relative_to_absolute (GNUNET_TIME_relative_ntoh (put->expiration));
3213
3214   /* first, validate! */
3215   switch (type)
3216     {
3217     case GNUNET_DATASTORE_BLOCKTYPE_DBLOCK:
3218     case GNUNET_DATASTORE_BLOCKTYPE_IBLOCK:
3219       GNUNET_CRYPTO_hash (&put[1], dsize, &query);
3220       break;
3221     case GNUNET_DATASTORE_BLOCKTYPE_KBLOCK:
3222       if (GNUNET_OK !=
3223           check_kblock ((const struct KBlock*) &put[1],
3224                         dsize,
3225                         &query))
3226         return GNUNET_SYSERR;
3227       break;
3228     case GNUNET_DATASTORE_BLOCKTYPE_SBLOCK:
3229       if (GNUNET_OK !=
3230           check_sblock ((const struct SBlock*) &put[1],
3231                         dsize,
3232                         &query,
3233                         &prq.namespace))
3234         return GNUNET_SYSERR;
3235       break;
3236     case GNUNET_DATASTORE_BLOCKTYPE_SKBLOCK:
3237       // FIXME -- validate SKBLOCK!
3238       GNUNET_break (0);
3239       return GNUNET_OK;
3240     default:
3241       /* unknown block type */
3242       GNUNET_break_op (0);
3243       return GNUNET_SYSERR;
3244     }
3245
3246   /* now, lookup 'query' */
3247   prq.data = (const void*) &put[1];
3248   prq.size = dsize;
3249   prq.type = type;
3250   prq.expiration = expiration;
3251   prq.priority = 0;
3252   GNUNET_CONTAINER_multihashmap_get_multiple (requests_by_query,
3253                                               &query,
3254                                               &process_reply,
3255                                               &prq);
3256   // FIXME: if migration is on and load is low,
3257   // queue to store data in datastore;
3258   // use "prq.priority" for that!
3259   return GNUNET_OK;
3260 }
3261
3262
3263 /**
3264  * List of handlers for P2P messages
3265  * that we care about.
3266  */
3267 static struct GNUNET_CORE_MessageHandler p2p_handlers[] =
3268   {
3269     { &handle_p2p_get,
3270       GNUNET_MESSAGE_TYPE_FS_GET, 0 },
3271     { &handle_p2p_put,
3272       GNUNET_MESSAGE_TYPE_FS_PUT, 0 },
3273     { NULL, 0, 0 }
3274   };
3275
3276
3277 /**
3278  * Task that will try to initiate a connection with the
3279  * core service.
3280  *
3281  * @param cls unused
3282  * @param tc unused
3283  */
3284 static void
3285 core_connect_task (void *cls,
3286                    const struct GNUNET_SCHEDULER_TaskContext *tc);
3287
3288
3289 /**
3290  * Function called by the core after we've
3291  * connected.
3292  *
3293  * @param cls closure, unused
3294  * @param server handle to the core service
3295  * @param my_identity our peer identity (unused)
3296  * @param publicKey our public key (unused)
3297  */
3298 static void
3299 core_start_cb (void *cls,
3300                struct GNUNET_CORE_Handle * server,
3301                const struct GNUNET_PeerIdentity *
3302                my_identity,
3303                const struct
3304                GNUNET_CRYPTO_RsaPublicKeyBinaryEncoded *
3305                publicKey)
3306 {
3307   if (server == NULL)
3308     {
3309       GNUNET_SCHEDULER_add_delayed (sched,
3310                                     GNUNET_TIME_UNIT_SECONDS,
3311                                     &core_connect_task,
3312                                     NULL);
3313       return;
3314     }
3315   core = server;
3316 }
3317
3318
3319 /**
3320  * Task that will try to initiate a connection with the
3321  * core service.
3322  *
3323  * @param cls unused
3324  * @param tc unused
3325  */
3326 static void
3327 core_connect_task (void *cls,
3328                    const struct GNUNET_SCHEDULER_TaskContext *tc)
3329 {
3330   GNUNET_CORE_connect (sched,
3331                        cfg,
3332                        GNUNET_TIME_UNIT_FOREVER_REL,
3333                        NULL,
3334                        &core_start_cb,
3335                        &peer_connect_handler,
3336                        &peer_disconnect_handler,
3337                        NULL,
3338                        NULL, GNUNET_NO,
3339                        NULL, GNUNET_NO,
3340                        p2p_handlers);
3341 }
3342
3343
3344 /**
3345  * Process fs requests.
3346  *
3347  * @param cls closure
3348  * @param s scheduler to use
3349  * @param server the initialized server
3350  * @param c configuration to use
3351  */
3352 static void
3353 run (void *cls,
3354      struct GNUNET_SCHEDULER_Handle *s,
3355      struct GNUNET_SERVER_Handle *server,
3356      const struct GNUNET_CONFIGURATION_Handle *c)
3357 {
3358   sched = s;
3359   cfg = c;
3360
3361   ifm = GNUNET_CONTAINER_multihashmap_create (128);
3362   requests_by_query = GNUNET_CONTAINER_multihashmap_create (128); // FIXME: get size from config
3363   requests_by_peer = GNUNET_CONTAINER_multihashmap_create (128); // FIXME: get size from config
3364   connected_peers = GNUNET_CONTAINER_multihashmap_create (64);
3365   requests_by_expiration = GNUNET_CONTAINER_heap_create (GNUNET_CONTAINER_HEAP_ORDER_MIN);
3366   read_index_list ();
3367   dsh = GNUNET_DATASTORE_connect (cfg,
3368                                   sched);
3369   if (NULL == dsh)
3370     {
3371       GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
3372                   _("Failed to connect to datastore service.\n"));
3373       return;
3374     }
3375   GNUNET_SERVER_disconnect_notify (server,
3376                                    &handle_client_disconnect,
3377                                    NULL);
3378   GNUNET_SERVER_add_handlers (server, handlers);
3379   core_connect_task (NULL, NULL);
3380   GNUNET_SCHEDULER_add_delayed (sched,
3381                                 GNUNET_TIME_UNIT_FOREVER_REL,
3382                                 &shutdown_task,
3383                                 NULL);
3384 }
3385
3386
3387 /**
3388  * The main function for the fs service.
3389  *
3390  * @param argc number of arguments from the command line
3391  * @param argv command line arguments
3392  * @return 0 ok, 1 on error
3393  */
3394 int
3395 main (int argc, char *const *argv)
3396 {
3397   return (GNUNET_OK ==
3398           GNUNET_SERVICE_run (argc,
3399                               argv,
3400                               "fs", &run, NULL)) ? 0 : 1;
3401 }
3402
3403 /* end of gnunet-service-fs.c */