src/fs/gnunet-service-fs.c

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