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   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 846               "Trying to migrate `%s' (%u bytes) to `%s'\n",
 847               GNUNET_h2s (&mb->query),
 848               msize,
 849               GNUNET_i2s (&cppid));
 850   cp->cth
 851     = GNUNET_CORE_notify_transmit_ready (core,
 852                                          0, GNUNET_TIME_UNIT_FOREVER_REL,
 853                                          (const struct GNUNET_PeerIdentity*) key,
 854                                          msize + sizeof (struct PutMessage),
 855                                          &transmit_to_peer,
 856                                          cp);
 857   return GNUNET_YES;
 858 }
 859
 860
 861 /**
 862  * Task that is run periodically to obtain blocks for content
 863  * migration
 864  *
 865  * @param cls unused
 866  * @param tc scheduler context (also unused)
 867  */
 868 static void
 869 gather_migration_blocks (void *cls,
 870                          const struct GNUNET_SCHEDULER_TaskContext *tc);
 871
 872
 873 /**
 874  * If the migration task is not currently running, consider
 875  * (re)scheduling it with the appropriate delay.
 876  */
 877 static void
 878 consider_migration_gathering ()
 879 {
 880   struct GNUNET_TIME_Relative delay;
 881
 882   if (mig_qe != NULL)
 883     return;
 884   if (mig_task != GNUNET_SCHEDULER_NO_TASK)
 885     return;
 886   delay = GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS,
 887                                          mig_size);
 888   delay = GNUNET_TIME_relative_divide (delay,
 889                                        MAX_MIGRATION_QUEUE);
 890   delay = GNUNET_TIME_relative_max (delay,
 891                                     min_migration_delay);
 892   mig_task = GNUNET_SCHEDULER_add_delayed (sched,
 893                                            delay,
 894                                            &gather_migration_blocks,
 895                                            NULL);
 896 }
 897
 898
 899 /**
 900  * Process content offered for migration.
 901  *
 902  * @param cls closure
 903  * @param key key for the content
 904  * @param size number of bytes in data
 905  * @param data content stored
 906  * @param type type of the content
 907  * @param priority priority of the content
 908  * @param anonymity anonymity-level for the content
 909  * @param expiration expiration time for the content
 910  * @param uid unique identifier for the datum;
 911  *        maybe 0 if no unique identifier is available
 912  */
 913 static void
 914 process_migration_content (void *cls,
 915                            const GNUNET_HashCode * key,
 916                            uint32_t size,
 917                            const void *data,
 918                            enum GNUNET_BLOCK_Type type,
 919                            uint32_t priority,
 920                            uint32_t anonymity,
 921                            struct GNUNET_TIME_Absolute
 922                            expiration, uint64_t uid)
 923 {
 924   struct MigrationReadyBlock *mb;
 925
 926   if (key == NULL)
 927     {
 928       mig_qe = NULL;
 929       if (mig_size < MAX_MIGRATION_QUEUE)
 930         consider_migration_gathering ();
 931       return;
 932     }
 933   if (type == GNUNET_BLOCK_TYPE_ONDEMAND)
 934     {
 935       if (GNUNET_OK !=
 936           GNUNET_FS_handle_on_demand_block (key, size, data,
 937                                             type, priority, anonymity,
 938                                             expiration, uid,
 939                                             &process_migration_content,
 940                                             NULL))
 941         GNUNET_DATASTORE_get_next (dsh, GNUNET_YES);
 942       return;
 943     }
 944 #if DEBUG_FS
 945   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 946               "Retrieved block `%s' of type %u for migration\n",
 947               GNUNET_h2s (key),
 948               type);
 949 #endif
 950   mb = GNUNET_malloc (sizeof (struct MigrationReadyBlock) + size);
 951   mb->query = *key;
 952   mb->expiration = expiration;
 953   mb->size = size;
 954   mb->type = type;
 955   memcpy (&mb[1], data, size);
 956   GNUNET_CONTAINER_DLL_insert_after (mig_head,
 957                                      mig_tail,
 958                                      mig_tail,
 959                                      mb);
 960   mig_size++;
 961   GNUNET_CONTAINER_multihashmap_iterate (connected_peers,
 962                                          &consider_migration,
 963                                          mb);
 964   GNUNET_DATASTORE_get_next (dsh, GNUNET_YES);
 965 }
 966
 967
 968 /**
 969  * Task that is run periodically to obtain blocks for content
 970  * migration
 971  *
 972  * @param cls unused
 973  * @param tc scheduler context (also unused)
 974  */
 975 static void
 976 gather_migration_blocks (void *cls,
 977                          const struct GNUNET_SCHEDULER_TaskContext *tc)
 978 {
 979   mig_task = GNUNET_SCHEDULER_NO_TASK;
 980   mig_qe = GNUNET_DATASTORE_get_random (dsh, 0, -1,
 981                                         GNUNET_TIME_UNIT_FOREVER_REL,
 982                                         &process_migration_content, NULL);
 983   GNUNET_assert (mig_qe != NULL);
 984 }
 985
 986
 987 /**
 988  * We're done with a particular message list entry.
 989  * Free all associated resources.
 990  *
 991  * @param pml entry to destroy
 992  */
 993 static void
 994 destroy_pending_message_list_entry (struct PendingMessageList *pml)
 995 {
 996   GNUNET_CONTAINER_DLL_remove (pml->req->pending_head,
 997                                pml->req->pending_tail,
 998                                pml);
 999   GNUNET_CONTAINER_DLL_remove (pml->target->pending_messages_head,
1000                                pml->target->pending_messages_tail,
1001                                pml->pm);
1002   pml->target->pending_requests--;
1003   GNUNET_free (pml->pm);
1004   GNUNET_free (pml);
1005 }
1006
1007
1008 /**
1009  * Destroy the given pending message (and call the respective
1010  * continuation).
1011  *
1012  * @param pm message to destroy
1013  * @param tpid id of peer that the message was delivered to, or 0 for none
1014  */
1015 static void
1016 destroy_pending_message (struct PendingMessage *pm,
1017                          GNUNET_PEER_Id tpid)
1018 {
1019   struct PendingMessageList *pml = pm->pml;
1020   TransmissionContinuation cont;
1021   void *cont_cls;
1022
1023   GNUNET_assert (pml->pm == pm);
1024   GNUNET_assert ( (tpid == 0) || (tpid == pml->target->pid) );
1025   cont = pm->cont;
1026   cont_cls = pm->cont_cls;
1027   destroy_pending_message_list_entry (pml);
1028   cont (cont_cls, tpid);
1029 }
1030
1031
1032 /**
1033  * We're done processing a particular request.
1034  * Free all associated resources.
1035  *
1036  * @param pr request to destroy
1037  */
1038 static void
1039 destroy_pending_request (struct PendingRequest *pr)
1040 {
1041   struct GNUNET_PeerIdentity pid;
1042
1043   if (pr->hnode != NULL)
1044     {
1045       GNUNET_CONTAINER_heap_remove_node (requests_by_expiration_heap,
1046                                          pr->hnode);
1047       pr->hnode = NULL;
1048     }
1049   if (NULL == pr->client_request_list)
1050     {
1051       GNUNET_STATISTICS_update (stats,
1052                                 gettext_noop ("# P2P searches active"),
1053                                 -1,
1054                                 GNUNET_NO);
1055     }
1056   else
1057     {
1058       GNUNET_STATISTICS_update (stats,
1059                                 gettext_noop ("# client searches active"),
1060                                 -1,
1061                                 GNUNET_NO);
1062     }
1063   /* might have already been removed from map in 'process_reply' (if
1064      there was a unique reply) or never inserted if it was a
1065      duplicate; hence ignore the return value here */
1066   (void) GNUNET_CONTAINER_multihashmap_remove (query_request_map,
1067                                                &pr->query,
1068                                                pr);
1069   if (pr->qe != NULL)
1070      {
1071       GNUNET_DATASTORE_cancel (pr->qe);
1072       pr->qe = NULL;
1073     }
1074   if (pr->client_request_list != NULL)
1075     {
1076       GNUNET_CONTAINER_DLL_remove (pr->client_request_list->client_list->rl_head,
1077                                    pr->client_request_list->client_list->rl_tail,
1078                                    pr->client_request_list);
1079       GNUNET_free (pr->client_request_list);
1080       pr->client_request_list = NULL;
1081     }
1082   if (pr->cp != NULL)
1083     {
1084       GNUNET_PEER_resolve (pr->cp->pid,
1085                            &pid);
1086       (void) GNUNET_CONTAINER_multihashmap_remove (peer_request_map,
1087                                                    &pid.hashPubKey,
1088                                                    pr);
1089       pr->cp = NULL;
1090     }
1091   if (pr->bf != NULL)
1092     {
1093       GNUNET_CONTAINER_bloomfilter_free (pr->bf);
1094       pr->bf = NULL;
1095     }
1096   if (pr->irc != NULL)
1097     {
1098       GNUNET_CORE_peer_change_preference_cancel (pr->irc);
1099       pr->irc = NULL;
1100     }
1101   if (pr->replies_seen != NULL)
1102     {
1103       GNUNET_free (pr->replies_seen);
1104       pr->replies_seen = NULL;
1105     }
1106   if (pr->task != GNUNET_SCHEDULER_NO_TASK)
1107     {
1108       GNUNET_SCHEDULER_cancel (sched,
1109                                pr->task);
1110       pr->task = GNUNET_SCHEDULER_NO_TASK;
1111     }
1112   while (NULL != pr->pending_head)
1113     destroy_pending_message_list_entry (pr->pending_head);
1114   GNUNET_PEER_change_rc (pr->target_pid, -1);
1115   if (pr->used_pids != NULL)
1116     {
1117       GNUNET_PEER_decrement_rcs (pr->used_pids, pr->used_pids_off);
1118       GNUNET_free (pr->used_pids);
1119       pr->used_pids_off = 0;
1120       pr->used_pids_size = 0;
1121       pr->used_pids = NULL;
1122     }
1123   GNUNET_free (pr);
1124 }
1125
1126
1127 /**
1128  * Method called whenever a given peer connects.
1129  *
1130  * @param cls closure, not used
1131  * @param peer peer identity this notification is about
1132  * @param latency reported latency of the connection with 'other'
1133  * @param distance reported distance (DV) to 'other'
1134  */
1135 static void
1136 peer_connect_handler (void *cls,
1137                       const struct
1138                       GNUNET_PeerIdentity * peer,
1139                       struct GNUNET_TIME_Relative latency,
1140                       uint32_t distance)
1141 {
1142   struct ConnectedPeer *cp;
1143   struct MigrationReadyBlock *pos;
1144
1145   cp = GNUNET_malloc (sizeof (struct ConnectedPeer));
1146   cp->pid = GNUNET_PEER_intern (peer);
1147   GNUNET_break (GNUNET_OK ==
1148                 GNUNET_CONTAINER_multihashmap_put (connected_peers,
1149                                                    &peer->hashPubKey,
1150                                                    cp,
1151                                                    GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
1152
1153   pos = mig_head;
1154   while (NULL != pos)
1155     {
1156       (void) consider_migration (pos, &peer->hashPubKey, cp);
1157       pos = pos->next;
1158     }
1159 }
1160
1161
1162
1163 /**
1164  * Free (each) request made by the peer.
1165  *
1166  * @param cls closure, points to peer that the request belongs to
1167  * @param key current key code
1168  * @param value value in the hash map
1169  * @return GNUNET_YES (we should continue to iterate)
1170  */
1171 static int
1172 destroy_request (void *cls,
1173                  const GNUNET_HashCode * key,
1174                  void *value)
1175 {
1176   const struct GNUNET_PeerIdentity * peer = cls;
1177   struct PendingRequest *pr = value;
1178
1179   GNUNET_break (GNUNET_YES ==
1180                 GNUNET_CONTAINER_multihashmap_remove (peer_request_map,
1181                                                       &peer->hashPubKey,
1182                                                       pr));
1183   destroy_pending_request (pr);
1184   return GNUNET_YES;
1185 }
1186
1187
1188 /**
1189  * Method called whenever a peer disconnects.
1190  *
1191  * @param cls closure, not used
1192  * @param peer peer identity this notification is about
1193  */
1194 static void
1195 peer_disconnect_handler (void *cls,
1196                          const struct
1197                          GNUNET_PeerIdentity * peer)
1198 {
1199   struct ConnectedPeer *cp;
1200   struct PendingMessage *pm;
1201   unsigned int i;
1202   struct MigrationReadyBlock *pos;
1203   struct MigrationReadyBlock *next;
1204
1205   GNUNET_CONTAINER_multihashmap_get_multiple (peer_request_map,
1206                                               &peer->hashPubKey,
1207                                               &destroy_request,
1208                                               (void*) peer);
1209   cp = GNUNET_CONTAINER_multihashmap_get (connected_peers,
1210                                           &peer->hashPubKey);
1211   if (cp == NULL)
1212     return;
1213   for (i=0;i<CS2P_SUCCESS_LIST_SIZE;i++)
1214     {
1215       if (NULL != cp->last_client_replies[i])
1216         {
1217           GNUNET_SERVER_client_drop (cp->last_client_replies[i]);
1218           cp->last_client_replies[i] = NULL;
1219         }
1220     }
1221   GNUNET_break (GNUNET_YES ==
1222                 GNUNET_CONTAINER_multihashmap_remove (connected_peers,
1223                                                       &peer->hashPubKey,
1224                                                       cp));
1225   /* remove this peer from migration considerations; schedule
1226      alternatives */
1227   next = mig_head;
1228   while (NULL != (pos = next))
1229     {
1230       next = pos->next;
1231       for (i=0;i<MIGRATION_LIST_SIZE;i++)
1232         {
1233           if (pos->target_list[i] == cp->pid)
1234             {
1235               GNUNET_PEER_change_rc (pos->target_list[i], -1);
1236               pos->target_list[i] = 0;
1237             }
1238          }
1239       if (pos->used_targets >= GNUNET_CONTAINER_multihashmap_size (connected_peers))
1240         {
1241           delete_migration_block (pos);
1242           consider_migration_gathering ();
1243           continue;
1244         }
1245       GNUNET_CONTAINER_multihashmap_iterate (connected_peers,
1246                                              &consider_migration,
1247                                              pos);
1248     }
1249
1250   GNUNET_PEER_change_rc (cp->pid, -1);
1251   GNUNET_PEER_decrement_rcs (cp->last_p2p_replies, P2P_SUCCESS_LIST_SIZE);
1252   if (NULL != cp->cth)
1253     GNUNET_CORE_notify_transmit_ready_cancel (cp->cth);
1254   while (NULL != (pm = cp->pending_messages_head))
1255     destroy_pending_message (pm, 0 /* delivery failed */);
1256   GNUNET_break (0 == cp->pending_requests);
1257   GNUNET_free (cp);
1258 }
1259
1260
1261 /**
1262  * Iterator over hash map entries that removes all occurences
1263  * of the given 'client' from the 'last_client_replies' of the
1264  * given connected peer.
1265  *
1266  * @param cls closure, the 'struct GNUNET_SERVER_Client*' to remove
1267  * @param key current key code (unused)
1268  * @param value value in the hash map (the 'struct ConnectedPeer*' to change)
1269  * @return GNUNET_YES (we should continue to iterate)
1270  */
1271 static int
1272 remove_client_from_last_client_replies (void *cls,
1273                                         const GNUNET_HashCode * key,
1274                                         void *value)
1275 {
1276   struct GNUNET_SERVER_Client *client = cls;
1277   struct ConnectedPeer *cp = value;
1278   unsigned int i;
1279
1280   for (i=0;i<CS2P_SUCCESS_LIST_SIZE;i++)
1281     {
1282       if (cp->last_client_replies[i] == client)
1283         {
1284           GNUNET_SERVER_client_drop (cp->last_client_replies[i]);
1285           cp->last_client_replies[i] = NULL;
1286         }
1287     }
1288   return GNUNET_YES;
1289 }
1290
1291
1292 /**
1293  * A client disconnected.  Remove all of its pending queries.
1294  *
1295  * @param cls closure, NULL
1296  * @param client identification of the client
1297  */
1298 static void
1299 handle_client_disconnect (void *cls,
1300                           struct GNUNET_SERVER_Client
1301                           * client)
1302 {
1303   struct ClientList *pos;
1304   struct ClientList *prev;
1305   struct ClientRequestList *rcl;
1306   struct ClientResponseMessage *creply;
1307
1308   if (client == NULL)
1309     return;
1310   prev = NULL;
1311   pos = client_list;
1312   while ( (NULL != pos) &&
1313           (pos->client != client) )
1314     {
1315       prev = pos;
1316       pos = pos->next;
1317     }
1318   if (pos == NULL)
1319     return; /* no requests pending for this client */
1320   while (NULL != (rcl = pos->rl_head))
1321     {
1322       GNUNET_log (GNUNET_ERROR_TYPE_INFO,
1323                   "Destroying pending request `%s' on disconnect\n",
1324                   GNUNET_h2s (&rcl->req->query));
1325       destroy_pending_request (rcl->req);
1326     }
1327   if (prev == NULL)
1328     client_list = pos->next;
1329   else
1330     prev->next = pos->next;
1331   if (pos->th != NULL)
1332     {
1333       GNUNET_CONNECTION_notify_transmit_ready_cancel (pos->th);
1334       pos->th = NULL;
1335     }
1336   while (NULL != (creply = pos->res_head))
1337     {
1338       GNUNET_CONTAINER_DLL_remove (pos->res_head,
1339                                    pos->res_tail,
1340                                    creply);
1341       GNUNET_free (creply);
1342     }
1343   GNUNET_SERVER_client_drop (pos->client);
1344   GNUNET_free (pos);
1345   GNUNET_CONTAINER_multihashmap_iterate (connected_peers,
1346                                          &remove_client_from_last_client_replies,
1347                                          client);
1348 }
1349
1350
1351 /**
1352  * Iterator to free peer entries.
1353  *
1354  * @param cls closure, unused
1355  * @param key current key code
1356  * @param value value in the hash map (peer entry)
1357  * @return GNUNET_YES (we should continue to iterate)
1358  */
1359 static int
1360 clean_peer (void *cls,
1361             const GNUNET_HashCode * key,
1362             void *value)
1363 {
1364   peer_disconnect_handler (NULL, (const struct GNUNET_PeerIdentity*) key);
1365   return GNUNET_YES;
1366 }
1367
1368
1369 /**
1370  * Task run during shutdown.
1371  *
1372  * @param cls unused
1373  * @param tc unused
1374  */
1375 static void
1376 shutdown_task (void *cls,
1377                const struct GNUNET_SCHEDULER_TaskContext *tc)
1378 {
1379   if (mig_qe != NULL)
1380     {
1381       GNUNET_DATASTORE_cancel (mig_qe);
1382       mig_qe = NULL;
1383     }
1384   if (GNUNET_SCHEDULER_NO_TASK != mig_task)
1385     {
1386       GNUNET_SCHEDULER_cancel (sched, mig_task);
1387       mig_task = GNUNET_SCHEDULER_NO_TASK;
1388     }
1389   while (client_list != NULL)
1390     handle_client_disconnect (NULL,
1391                               client_list->client);
1392   GNUNET_CONTAINER_multihashmap_iterate (connected_peers,
1393                                          &clean_peer,
1394                                          NULL);
1395   GNUNET_break (0 == GNUNET_CONTAINER_heap_get_size (requests_by_expiration_heap));
1396   GNUNET_CONTAINER_heap_destroy (requests_by_expiration_heap);
1397   requests_by_expiration_heap = 0;
1398   GNUNET_CONTAINER_multihashmap_destroy (connected_peers);
1399   connected_peers = NULL;
1400   GNUNET_break (0 == GNUNET_CONTAINER_multihashmap_size (query_request_map));
1401   GNUNET_CONTAINER_multihashmap_destroy (query_request_map);
1402   query_request_map = NULL;
1403   GNUNET_break (0 == GNUNET_CONTAINER_multihashmap_size (peer_request_map));
1404   GNUNET_CONTAINER_multihashmap_destroy (peer_request_map);
1405   peer_request_map = NULL;
1406   GNUNET_assert (NULL != core);
1407   GNUNET_CORE_disconnect (core);
1408   core = NULL;
1409   if (stats != NULL)
1410     {
1411       GNUNET_STATISTICS_destroy (stats, GNUNET_NO);
1412       stats = NULL;
1413     }
1414   GNUNET_DATASTORE_disconnect (dsh,
1415                                GNUNET_NO);
1416   while (mig_head != NULL)
1417     delete_migration_block (mig_head);
1418   GNUNET_assert (0 == mig_size);
1419   dsh = NULL;
1420   sched = NULL;
1421   cfg = NULL;
1422 }
1423
1424
1425 /* ******************* Utility functions  ******************** */
1426
1427
1428 /**
1429  * Transmit messages by copying it to the target buffer
1430  * "buf".  "buf" will be NULL and "size" zero if the socket was closed
1431  * for writing in the meantime.  In that case, do nothing
1432  * (the disconnect or shutdown handler will take care of the rest).
1433  * If we were able to transmit messages and there are still more
1434  * pending, ask core again for further calls to this function.
1435  *
1436  * @param cls closure, pointer to the 'struct ConnectedPeer*'
1437  * @param size number of bytes available in buf
1438  * @param buf where the callee should write the message
1439  * @return number of bytes written to buf
1440  */
1441 static size_t
1442 transmit_to_peer (void *cls,
1443                   size_t size, void *buf)
1444 {
1445   struct ConnectedPeer *cp = cls;
1446   char *cbuf = buf;
1447   struct GNUNET_PeerIdentity pid;
1448   struct PendingMessage *pm;
1449   struct MigrationReadyBlock *mb;
1450   struct MigrationReadyBlock *next;
1451   struct PutMessage migm;
1452   size_t msize;
1453   unsigned int i;
1454
1455   cp->cth = NULL;
1456   if (NULL == buf)
1457     {
1458 #if DEBUG_FS
1459       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1460                   "Dropping message, core too busy.\n");
1461 #endif
1462       return 0;
1463     }
1464   msize = 0;
1465   while ( (NULL != (pm = cp->pending_messages_head) ) &&
1466           (pm->msize <= size) )
1467     {
1468       memcpy (&cbuf[msize], &pm[1], pm->msize);
1469       msize += pm->msize;
1470       size -= pm->msize;
1471       destroy_pending_message (pm, cp->pid);
1472     }
1473   if (NULL != pm)
1474     {
1475       GNUNET_PEER_resolve (cp->pid,
1476                            &pid);
1477       cp->cth = GNUNET_CORE_notify_transmit_ready (core,
1478                                                    pm->priority,
1479                                                    GNUNET_CONSTANTS_SERVICE_TIMEOUT,
1480                                                    &pid,
1481                                                    pm->msize,
1482                                                    &transmit_to_peer,
1483                                                    cp);
1484     }
1485   else
1486     {
1487       next = mig_head;
1488       while (NULL != (mb = next))
1489         {
1490           next = mb->next;
1491           for (i=0;i<MIGRATION_LIST_SIZE;i++)
1492             {
1493               if ( (cp->pid == mb->target_list[i]) &&
1494                    (mb->size + sizeof (migm) <= size) )
1495                 {
1496                   GNUNET_PEER_change_rc (mb->target_list[i], -1);
1497                   mb->target_list[i] = 0;
1498                   mb->used_targets++;
1499                   migm.header.size = htons (sizeof (migm) + mb->size);
1500                   migm.header.type = htons (GNUNET_MESSAGE_TYPE_FS_PUT);
1501                   migm.type = htonl (mb->type);
1502                   migm.expiration = GNUNET_TIME_absolute_hton (mb->expiration);
1503                   memcpy (&cbuf[msize], &migm, sizeof (migm));
1504                   msize += sizeof (migm);
1505                   size -= sizeof (migm);
1506                   memcpy (&cbuf[msize], &mb[1], mb->size);
1507                   msize += mb->size;
1508                   size -= mb->size;
1509 #if DEBUG_FS
1510                   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1511                               "Pushing migration block `%s' (%u bytes) to `%s'\n",
1512                               GNUNET_h2s (&mb->query),
1513                               mb->size,
1514                               GNUNET_i2s (&pid));
1515 #endif
1516                   break;
1517                 }
1518               else
1519                 {
1520 #if DEBUG_FS
1521                   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1522                               "Migration block `%s' (%u bytes) is not on migration list for peer `%s'\n",
1523                               GNUNET_h2s (&mb->query),
1524                               mb->size,
1525                               GNUNET_i2s (&pid));
1526 #endif
1527                 }
1528             }
1529           if ( (mb->used_targets >= MIGRATION_TARGET_COUNT) ||
1530                (mb->used_targets >= GNUNET_CONTAINER_multihashmap_size (connected_peers)) )
1531             {
1532               delete_migration_block (mb);
1533               consider_migration_gathering ();
1534             }
1535         }
1536       consider_migration (NULL,
1537                           &pid.hashPubKey,
1538                           cp);
1539     }
1540 #if DEBUG_FS
1541   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1542               "Transmitting %u bytes to peer %u\n",
1543               msize,
1544               cp->pid);
1545 #endif
1546   return msize;
1547 }
1548
1549
1550 /**
1551  * Add a message to the set of pending messages for the given peer.
1552  *
1553  * @param cp peer to send message to
1554  * @param pm message to queue
1555  * @param pr request on which behalf this message is being queued
1556  */
1557 static void
1558 add_to_pending_messages_for_peer (struct ConnectedPeer *cp,
1559                                   struct PendingMessage *pm,
1560                                   struct PendingRequest *pr)
1561 {
1562   struct PendingMessage *pos;
1563   struct PendingMessageList *pml;
1564   struct GNUNET_PeerIdentity pid;
1565
1566   GNUNET_assert (pm->next == NULL);
1567   GNUNET_assert (pm->pml == NULL);
1568   pml = GNUNET_malloc (sizeof (struct PendingMessageList));
1569   pml->req = pr;
1570   pml->target = cp;
1571   pml->pm = pm;
1572   pm->pml = pml;
1573   GNUNET_CONTAINER_DLL_insert (pr->pending_head,
1574                                pr->pending_tail,
1575                                pml);
1576   pos = cp->pending_messages_head;
1577   while ( (pos != NULL) &&
1578           (pm->priority < pos->priority) )
1579     pos = pos->next;
1580   GNUNET_CONTAINER_DLL_insert_after (cp->pending_messages_head,
1581                                      cp->pending_messages_tail,
1582                                      pos,
1583                                      pm);
1584   cp->pending_requests++;
1585   if (cp->pending_requests > MAX_QUEUE_PER_PEER)
1586     destroy_pending_message (cp->pending_messages_tail, 0);
1587   GNUNET_PEER_resolve (cp->pid, &pid);
1588   if (NULL != cp->cth)
1589     GNUNET_CORE_notify_transmit_ready_cancel (cp->cth);
1590   /* need to schedule transmission */
1591   cp->cth = GNUNET_CORE_notify_transmit_ready (core,
1592                                                cp->pending_messages_head->priority,
1593                                                MAX_TRANSMIT_DELAY,
1594                                                &pid,
1595                                                cp->pending_messages_head->msize,
1596                                                &transmit_to_peer,
1597                                                cp);
1598   if (cp->cth == NULL)
1599     {
1600 #if DEBUG_FS
1601       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1602                   "Failed to schedule transmission with core!\n");
1603 #endif
1604       GNUNET_STATISTICS_update (stats,
1605                                 gettext_noop ("# CORE transmission failures"),
1606                                 1,
1607                                 GNUNET_NO);
1608     }
1609 }
1610
1611
1612 /**
1613  * Mingle hash with the mingle_number to produce different bits.
1614  */
1615 static void
1616 mingle_hash (const GNUNET_HashCode * in,
1617              int32_t mingle_number,
1618              GNUNET_HashCode * hc)
1619 {
1620   GNUNET_HashCode m;
1621
1622   GNUNET_CRYPTO_hash (&mingle_number,
1623                       sizeof (int32_t),
1624                       &m);
1625   GNUNET_CRYPTO_hash_xor (&m, in, hc);
1626 }
1627
1628
1629 /**
1630  * Test if the load on this peer is too high
1631  * to even consider processing the query at
1632  * all.
1633  *
1634  * @return GNUNET_YES if the load is too high, GNUNET_NO otherwise
1635  */
1636 static int
1637 test_load_too_high ()
1638 {
1639   return GNUNET_NO; // FIXME
1640 }
1641
1642
1643 /* ******************* Pending Request Refresh Task ******************** */
1644
1645
1646
1647 /**
1648  * We use a random delay to make the timing of requests less
1649  * predictable.  This function returns such a random delay.  We add a base
1650  * delay of MAX_CORK_DELAY (1s).
1651  *
1652  * FIXME: make schedule dependent on the specifics of the request?
1653  * Or bandwidth and number of connected peers and load?
1654  *
1655  * @return random delay to use for some request, between 1s and 1000+TTL_DECREMENT ms
1656  */
1657 static struct GNUNET_TIME_Relative
1658 get_processing_delay ()
1659 {
1660   return
1661     GNUNET_TIME_relative_add (GNUNET_CONSTANTS_MAX_CORK_DELAY,
1662                               GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MILLISECONDS,
1663                                                              GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
1664                                                                                        TTL_DECREMENT)));
1665 }
1666
1667
1668 /**
1669  * We're processing a GET request from another peer and have decided
1670  * to forward it to other peers.  This function is called periodically
1671  * and should forward the request to other peers until we have all
1672  * possible replies.  If we have transmitted the *only* reply to
1673  * the initiator we should destroy the pending request.  If we have
1674  * many replies in the queue to the initiator, we should delay sending
1675  * out more queries until the reply queue has shrunk some.
1676  *
1677  * @param cls our "struct ProcessGetContext *"
1678  * @param tc unused
1679  */
1680 static void
1681 forward_request_task (void *cls,
1682                       const struct GNUNET_SCHEDULER_TaskContext *tc);
1683
1684
1685 /**
1686  * Function called after we either failed or succeeded
1687  * at transmitting a query to a peer.
1688  *
1689  * @param cls the requests "struct PendingRequest*"
1690  * @param tpid ID of receiving peer, 0 on transmission error
1691  */
1692 static void
1693 transmit_query_continuation (void *cls,
1694                              GNUNET_PEER_Id tpid)
1695 {
1696   struct PendingRequest *pr = cls;
1697
1698   GNUNET_STATISTICS_update (stats,
1699                             gettext_noop ("# queries scheduled for forwarding"),
1700                             -1,
1701                             GNUNET_NO);
1702   if (tpid == 0)
1703     {
1704 #if DEBUG_FS
1705       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1706                   "Transmission of request failed, will try again later.\n");
1707 #endif
1708       if (pr->task == GNUNET_SCHEDULER_NO_TASK)
1709         pr->task = GNUNET_SCHEDULER_add_delayed (sched,
1710                                                  get_processing_delay (),
1711                                                  &forward_request_task,
1712                                                  pr);
1713       return;
1714     }
1715   GNUNET_STATISTICS_update (stats,
1716                             gettext_noop ("# queries forwarded"),
1717                             1,
1718                             GNUNET_NO);
1719   GNUNET_PEER_change_rc (tpid, 1);
1720   if (pr->used_pids_off == pr->used_pids_size)
1721     GNUNET_array_grow (pr->used_pids,
1722                        pr->used_pids_size,
1723                        pr->used_pids_size * 2 + 2);
1724   pr->used_pids[pr->used_pids_off++] = tpid;
1725   if (pr->task == GNUNET_SCHEDULER_NO_TASK)
1726     pr->task = GNUNET_SCHEDULER_add_delayed (sched,
1727                                              get_processing_delay (),
1728                                              &forward_request_task,
1729                                              pr);
1730 }
1731
1732
1733 /**
1734  * How many bytes should a bloomfilter be if we have already seen
1735  * entry_count responses?  Note that BLOOMFILTER_K gives us the number
1736  * of bits set per entry.  Furthermore, we should not re-size the
1737  * filter too often (to keep it cheap).
1738  *
1739  * Since other peers will also add entries but not resize the filter,
1740  * we should generally pick a slightly larger size than what the
1741  * strict math would suggest.
1742  *
1743  * @return must be a power of two and smaller or equal to 2^15.
1744  */
1745 static size_t
1746 compute_bloomfilter_size (unsigned int entry_count)
1747 {
1748   size_t size;
1749   unsigned int ideal = (entry_count * BLOOMFILTER_K) / 4;
1750   uint16_t max = 1 << 15;
1751
1752   if (entry_count > max)
1753     return max;
1754   size = 8;
1755   while ((size < max) && (size < ideal))
1756     size *= 2;
1757   if (size > max)
1758     return max;
1759   return size;
1760 }
1761
1762
1763 /**
1764  * Recalculate our bloom filter for filtering replies.  This function
1765  * will create a new bloom filter from scratch, so it should only be
1766  * called if we have no bloomfilter at all (and hence can create a
1767  * fresh one of minimal size without problems) OR if our peer is the
1768  * initiator (in which case we may resize to larger than mimimum size).
1769  *
1770  * @param pr request for which the BF is to be recomputed
1771  */
1772 static void
1773 refresh_bloomfilter (struct PendingRequest *pr)
1774 {
1775   unsigned int i;
1776   size_t nsize;
1777   GNUNET_HashCode mhash;
1778
1779   nsize = compute_bloomfilter_size (pr->replies_seen_off);
1780   if (nsize == pr->bf_size)
1781     return; /* size not changed */
1782   if (pr->bf != NULL)
1783     GNUNET_CONTAINER_bloomfilter_free (pr->bf);
1784   pr->bf_size = nsize;
1785   pr->mingle = GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK, -1);
1786   pr->bf = GNUNET_CONTAINER_bloomfilter_init (NULL,
1787                                               pr->bf_size,
1788                                               BLOOMFILTER_K);
1789   for (i=0;i<pr->replies_seen_off;i++)
1790     {
1791       mingle_hash (&pr->replies_seen[i], pr->mingle, &mhash);
1792       GNUNET_CONTAINER_bloomfilter_add (pr->bf, &mhash);
1793     }
1794 }
1795
1796
1797 /**
1798  * Function called after we've tried to reserve a certain amount of
1799  * bandwidth for a reply.  Check if we succeeded and if so send our
1800  * query.
1801  *
1802  * @param cls the requests "struct PendingRequest*"
1803  * @param peer identifies the peer
1804  * @param bpm_in set to the current bandwidth limit (receiving) for this peer
1805  * @param bpm_out set to the current bandwidth limit (sending) for this peer
1806  * @param amount set to the amount that was actually reserved or unreserved
1807  * @param preference current traffic preference for the given peer
1808  */
1809 static void
1810 target_reservation_cb (void *cls,
1811                        const struct
1812                        GNUNET_PeerIdentity * peer,
1813                        struct GNUNET_BANDWIDTH_Value32NBO bpm_in,
1814                        struct GNUNET_BANDWIDTH_Value32NBO bpm_out,
1815                        int amount,
1816                        uint64_t preference)
1817 {
1818   struct PendingRequest *pr = cls;
1819   struct ConnectedPeer *cp;
1820   struct PendingMessage *pm;
1821   struct GetMessage *gm;
1822   GNUNET_HashCode *ext;
1823   char *bfdata;
1824   size_t msize;
1825   unsigned int k;
1826   int no_route;
1827   uint32_t bm;
1828
1829   pr->irc = NULL;
1830   if (peer == NULL)
1831     {
1832       /* error in communication with core, try again later */
1833       if (pr->task == GNUNET_SCHEDULER_NO_TASK)
1834         pr->task = GNUNET_SCHEDULER_add_delayed (sched,
1835                                                  get_processing_delay (),
1836                                                  &forward_request_task,
1837                                                  pr);
1838       return;
1839     }
1840   // (3) transmit, update ttl/priority
1841   cp = GNUNET_CONTAINER_multihashmap_get (connected_peers,
1842                                           &peer->hashPubKey);
1843   if (cp == NULL)
1844     {
1845       /* Peer must have just left */
1846 #if DEBUG_FS
1847       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1848                   "Selected peer disconnected!\n");
1849 #endif
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;
1856     }
1857   no_route = GNUNET_NO;
1858   /* FIXME: check against DBLOCK_SIZE and possibly return
1859      amount to reserve; however, this also needs to work
1860      with testcases which currently start out with a far
1861      too low per-peer bw limit, so they would never send
1862      anything.  Big issue. */
1863   if (amount == 0)
1864     {
1865       if (pr->cp == NULL)
1866         {
1867 #if DEBUG_FS > 1
1868           GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1869                       "Failed to reserve bandwidth for reply (got %d/%u bytes only)!\n",
1870                       amount,
1871                       DBLOCK_SIZE);
1872 #endif
1873           GNUNET_STATISTICS_update (stats,
1874                                     gettext_noop ("# reply bandwidth reservation requests failed"),
1875                                     1,
1876                                     GNUNET_NO);
1877           if (pr->task == GNUNET_SCHEDULER_NO_TASK)
1878             pr->task = GNUNET_SCHEDULER_add_delayed (sched,
1879                                                      get_processing_delay (),
1880                                                      &forward_request_task,
1881                                                      pr);
1882           return;  /* this target round failed */
1883         }
1884       /* FIXME: if we are "quite" busy, we may still want to skip
1885          this round; need more load detection code! */
1886       no_route = GNUNET_YES;
1887     }
1888
1889   GNUNET_STATISTICS_update (stats,
1890                             gettext_noop ("# queries scheduled for forwarding"),
1891                             1,
1892                             GNUNET_NO);
1893   /* build message and insert message into priority queue */
1894 #if DEBUG_FS
1895   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1896               "Forwarding request `%s' to `%4s'!\n",
1897               GNUNET_h2s (&pr->query),
1898               GNUNET_i2s (peer));
1899 #endif
1900   k = 0;
1901   bm = 0;
1902   if (GNUNET_YES == no_route)
1903     {
1904       bm |= GET_MESSAGE_BIT_RETURN_TO;
1905       k++;
1906     }
1907   if (pr->namespace != NULL)
1908     {
1909       bm |= GET_MESSAGE_BIT_SKS_NAMESPACE;
1910       k++;
1911     }
1912   if (pr->target_pid != 0)
1913     {
1914       bm |= GET_MESSAGE_BIT_TRANSMIT_TO;
1915       k++;
1916     }
1917   msize = sizeof (struct GetMessage) + pr->bf_size + k * sizeof(GNUNET_HashCode);
1918   GNUNET_assert (msize < GNUNET_SERVER_MAX_MESSAGE_SIZE);
1919   pm = GNUNET_malloc (sizeof (struct PendingMessage) + msize);
1920   pm->msize = msize;
1921   gm = (struct GetMessage*) &pm[1];
1922   gm->header.type = htons (GNUNET_MESSAGE_TYPE_FS_GET);
1923   gm->header.size = htons (msize);
1924   gm->type = htonl (pr->type);
1925   pr->remaining_priority /= 2;
1926   gm->priority = htonl (pr->remaining_priority);
1927   gm->ttl = htonl (pr->ttl);
1928   gm->filter_mutator = htonl(pr->mingle);
1929   gm->hash_bitmap = htonl (bm);
1930   gm->query = pr->query;
1931   ext = (GNUNET_HashCode*) &gm[1];
1932   k = 0;
1933   if (GNUNET_YES == no_route)
1934     GNUNET_PEER_resolve (pr->cp->pid, (struct GNUNET_PeerIdentity*) &ext[k++]);
1935   if (pr->namespace != NULL)
1936     memcpy (&ext[k++], pr->namespace, sizeof (GNUNET_HashCode));
1937   if (pr->target_pid != 0)
1938     GNUNET_PEER_resolve (pr->target_pid, (struct GNUNET_PeerIdentity*) &ext[k++]);
1939   bfdata = (char *) &ext[k];
1940   if (pr->bf != NULL)
1941     GNUNET_CONTAINER_bloomfilter_get_raw_data (pr->bf,
1942                                                bfdata,
1943                                                pr->bf_size);
1944   pm->cont = &transmit_query_continuation;
1945   pm->cont_cls = pr;
1946   add_to_pending_messages_for_peer (cp, pm, pr);
1947 }
1948
1949
1950 /**
1951  * Closure used for "target_peer_select_cb".
1952  */
1953 struct PeerSelectionContext
1954 {
1955   /**
1956    * The request for which we are selecting
1957    * peers.
1958    */
1959   struct PendingRequest *pr;
1960
1961   /**
1962    * Current "prime" target.
1963    */
1964   struct GNUNET_PeerIdentity target;
1965
1966   /**
1967    * How much do we like this target?
1968    */
1969   double target_score;
1970
1971 };
1972
1973
1974 /**
1975  * Function called for each connected peer to determine
1976  * which one(s) would make good targets for forwarding.
1977  *
1978  * @param cls closure (struct PeerSelectionContext)
1979  * @param key current key code (peer identity)
1980  * @param value value in the hash map (struct ConnectedPeer)
1981  * @return GNUNET_YES if we should continue to
1982  *         iterate,
1983  *         GNUNET_NO if not.
1984  */
1985 static int
1986 target_peer_select_cb (void *cls,
1987                        const GNUNET_HashCode * key,
1988                        void *value)
1989 {
1990   struct PeerSelectionContext *psc = cls;
1991   struct ConnectedPeer *cp = value;
1992   struct PendingRequest *pr = psc->pr;
1993   double score;
1994   unsigned int i;
1995   unsigned int pc;
1996
1997   /* 1) check that this peer is not the initiator */
1998   if (cp == pr->cp)
1999     return GNUNET_YES; /* skip */
2000
2001   /* 2) check if we have already (recently) forwarded to this peer */
2002   pc = 0;
2003   for (i=0;i<pr->used_pids_off;i++)
2004     if (pr->used_pids[i] == cp->pid)
2005       {
2006         pc++;
2007         if (0 != GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
2008                                            RETRY_PROBABILITY_INV))
2009           {
2010 #if DEBUG_FS
2011             GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2012                         "NOT re-trying query that was previously transmitted %u times\n",
2013                         (unsigned int) pr->used_pids_off);
2014 #endif
2015             return GNUNET_YES; /* skip */
2016           }
2017       }
2018 #if DEBUG_FS
2019   if (0 < pc)
2020     GNUNET_log (GNUNET_ERROR_TYPE_INFO,
2021                 "Re-trying query that was previously transmitted %u times to this peer\n",
2022                 (unsigned int) pc);
2023 #endif
2024   // 3) calculate how much we'd like to forward to this peer
2025   score = 42; // FIXME!
2026   // FIXME: also need API to gather data on responsiveness
2027   // of this peer (we have fields for that in 'cp', but
2028   // they are never set!)
2029
2030   /* store best-fit in closure */
2031   if (score > psc->target_score)
2032     {
2033       psc->target_score = score;
2034       psc->target.hashPubKey = *key;
2035     }
2036   return GNUNET_YES;
2037 }
2038
2039
2040 /**
2041  * The priority level imposes a bound on the maximum
2042  * value for the ttl that can be requested.
2043  *
2044  * @param ttl_in requested ttl
2045  * @param prio given priority
2046  * @return ttl_in if ttl_in is below the limit,
2047  *         otherwise the ttl-limit for the given priority
2048  */
2049 static int32_t
2050 bound_ttl (int32_t ttl_in, uint32_t prio)
2051 {
2052   unsigned long long allowed;
2053
2054   if (ttl_in <= 0)
2055     return ttl_in;
2056   allowed = ((unsigned long long) prio) * TTL_DECREMENT / 1000;
2057   if (ttl_in > allowed)
2058     {
2059       if (allowed >= (1 << 30))
2060         return 1 << 30;
2061       return allowed;
2062     }
2063   return ttl_in;
2064 }
2065
2066
2067 /**
2068  * We're processing a GET request from another peer and have decided
2069  * to forward it to other peers.  This function is called periodically
2070  * and should forward the request to other peers until we have all
2071  * possible replies.  If we have transmitted the *only* reply to
2072  * the initiator we should destroy the pending request.  If we have
2073  * many replies in the queue to the initiator, we should delay sending
2074  * out more queries until the reply queue has shrunk some.
2075  *
2076  * @param cls our "struct ProcessGetContext *"
2077  * @param tc unused
2078  */
2079 static void
2080 forward_request_task (void *cls,
2081                      const struct GNUNET_SCHEDULER_TaskContext *tc)
2082 {
2083   struct PendingRequest *pr = cls;
2084   struct PeerSelectionContext psc;
2085   struct ConnectedPeer *cp;
2086   struct GNUNET_TIME_Relative delay;
2087
2088   pr->task = GNUNET_SCHEDULER_NO_TASK;
2089   if (pr->irc != NULL)
2090     {
2091 #if DEBUG_FS
2092       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2093                   "Forwarding of query `%s' not attempted due to pending local lookup!\n",
2094                   GNUNET_h2s (&pr->query));
2095 #endif
2096       return; /* already pending */
2097     }
2098   if (GNUNET_YES == pr->local_only)
2099     return; /* configured to not do P2P search */
2100   /* (1) select target */
2101   psc.pr = pr;
2102   psc.target_score = DBL_MIN;
2103   GNUNET_CONTAINER_multihashmap_iterate (connected_peers,
2104                                          &target_peer_select_cb,
2105                                          &psc);
2106   if (psc.target_score == DBL_MIN)
2107     {
2108       delay = get_processing_delay ();
2109 #if DEBUG_FS
2110       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2111                   "No peer selected for forwarding of query `%s', will try again in %llu ms!\n",
2112                   GNUNET_h2s (&pr->query),
2113                   delay.value);
2114 #endif
2115       pr->task = GNUNET_SCHEDULER_add_delayed (sched,
2116                                                delay,
2117                                                &forward_request_task,
2118                                                pr);
2119       return; /* nobody selected */
2120     }
2121   /* (3) update TTL/priority */
2122
2123   if (pr->client_request_list != NULL)
2124     {
2125       /* FIXME: use better algorithm!? */
2126       if (0 == GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
2127                                          4))
2128         pr->priority++;
2129       /* FIXME: bound priority by "customary" priority used by other peers
2130          at this time! */
2131       pr->ttl = bound_ttl (pr->ttl + TTL_DECREMENT * 2,
2132                            pr->priority);
2133 #if DEBUG_FS
2134       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2135                   "Trying query `%s' with priority %u and TTL %d.\n",
2136                   GNUNET_h2s (&pr->query),
2137                   pr->priority,
2138                   pr->ttl);
2139 #endif
2140     }
2141   else
2142     {
2143       /* FIXME: should we do something here as well!? */
2144     }
2145
2146   /* (3) reserve reply bandwidth */
2147   cp = GNUNET_CONTAINER_multihashmap_get (connected_peers,
2148                                           &psc.target.hashPubKey);
2149   GNUNET_assert (NULL != cp);
2150   pr->irc = GNUNET_CORE_peer_change_preference (sched, cfg,
2151                                                 &psc.target,
2152                                                 GNUNET_CONSTANTS_SERVICE_TIMEOUT,
2153                                                 GNUNET_BANDWIDTH_value_init (UINT32_MAX),
2154                                                 DBLOCK_SIZE * 2,
2155                                                 (uint64_t) cp->inc_preference,
2156                                                 &target_reservation_cb,
2157                                                 pr);
2158   cp->inc_preference = 0.0;
2159 }
2160
2161
2162 /* **************************** P2P PUT Handling ************************ */
2163
2164
2165 /**
2166  * Function called after we either failed or succeeded
2167  * at transmitting a reply to a peer.
2168  *
2169  * @param cls the requests "struct PendingRequest*"
2170  * @param tpid ID of receiving peer, 0 on transmission error
2171  */
2172 static void
2173 transmit_reply_continuation (void *cls,
2174                              GNUNET_PEER_Id tpid)
2175 {
2176   struct PendingRequest *pr = cls;
2177
2178   switch (pr->type)
2179     {
2180     case GNUNET_BLOCK_TYPE_DBLOCK:
2181     case GNUNET_BLOCK_TYPE_IBLOCK:
2182       /* only one reply expected, done with the request! */
2183       destroy_pending_request (pr);
2184       break;
2185     case GNUNET_BLOCK_TYPE_ANY:
2186     case GNUNET_BLOCK_TYPE_KBLOCK:
2187     case GNUNET_BLOCK_TYPE_SBLOCK:
2188       break;
2189     default:
2190       GNUNET_break (0);
2191       break;
2192     }
2193 }
2194
2195
2196 /**
2197  * Transmit the given message by copying it to the target buffer
2198  * "buf".  "buf" will be NULL and "size" zero if the socket was closed
2199  * for writing in the meantime.  In that case, do nothing
2200  * (the disconnect or shutdown handler will take care of the rest).
2201  * If we were able to transmit messages and there are still more
2202  * pending, ask core again for further calls to this function.
2203  *
2204  * @param cls closure, pointer to the 'struct ClientList*'
2205  * @param size number of bytes available in buf
2206  * @param buf where the callee should write the message
2207  * @return number of bytes written to buf
2208  */
2209 static size_t
2210 transmit_to_client (void *cls,
2211                   size_t size, void *buf)
2212 {
2213   struct ClientList *cl = cls;
2214   char *cbuf = buf;
2215   struct ClientResponseMessage *creply;
2216   size_t msize;
2217
2218   cl->th = NULL;
2219   if (NULL == buf)
2220     {
2221 #if DEBUG_FS
2222       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2223                   "Not sending reply, client communication problem.\n");
2224 #endif
2225       return 0;
2226     }
2227   msize = 0;
2228   while ( (NULL != (creply = cl->res_head) ) &&
2229           (creply->msize <= size) )
2230     {
2231       memcpy (&cbuf[msize], &creply[1], creply->msize);
2232       msize += creply->msize;
2233       size -= creply->msize;
2234       GNUNET_CONTAINER_DLL_remove (cl->res_head,
2235                                    cl->res_tail,
2236                                    creply);
2237       GNUNET_free (creply);
2238     }
2239   if (NULL != creply)
2240     cl->th = GNUNET_SERVER_notify_transmit_ready (cl->client,
2241                                                   creply->msize,
2242                                                   GNUNET_TIME_UNIT_FOREVER_REL,
2243                                                   &transmit_to_client,
2244                                                   cl);
2245 #if DEBUG_FS
2246   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2247               "Transmitted %u bytes to client\n",
2248               (unsigned int) msize);
2249 #endif
2250   return msize;
2251 }
2252
2253
2254 /**
2255  * Closure for "process_reply" function.
2256  */
2257 struct ProcessReplyClosure
2258 {
2259   /**
2260    * The data for the reply.
2261    */
2262   const void *data;
2263
2264   /**
2265    * Who gave us this reply? NULL for local host.
2266    */
2267   struct ConnectedPeer *sender;
2268
2269   /**
2270    * When the reply expires.
2271    */
2272   struct GNUNET_TIME_Absolute expiration;
2273
2274   /**
2275    * Size of data.
2276    */
2277   size_t size;
2278
2279   /**
2280    * Namespace that this reply belongs to
2281    * (if it is of type SBLOCK).
2282    */
2283   GNUNET_HashCode namespace;
2284
2285   /**
2286    * Type of the block.
2287    */
2288   enum GNUNET_BLOCK_Type type;
2289
2290   /**
2291    * How much was this reply worth to us?
2292    */
2293   uint32_t priority;
2294
2295   /**
2296    * Did we finish processing the associated request?
2297    */
2298   int finished;
2299 };
2300
2301
2302 /**
2303  * We have received a reply; handle it!
2304  *
2305  * @param cls response (struct ProcessReplyClosure)
2306  * @param key our query
2307  * @param value value in the hash map (info about the query)
2308  * @return GNUNET_YES (we should continue to iterate)
2309  */
2310 static int
2311 process_reply (void *cls,
2312                const GNUNET_HashCode * key,
2313                void *value)
2314 {
2315   struct ProcessReplyClosure *prq = cls;
2316   struct PendingRequest *pr = value;
2317   struct PendingMessage *reply;
2318   struct ClientResponseMessage *creply;
2319   struct ClientList *cl;
2320   struct PutMessage *pm;
2321   struct ConnectedPeer *cp;
2322   struct GNUNET_TIME_Relative cur_delay;
2323   GNUNET_HashCode chash;
2324   GNUNET_HashCode mhash;
2325   size_t msize;
2326
2327 #if DEBUG_FS
2328   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2329               "Matched result (type %u) for query `%s' with pending request\n",
2330               (unsigned int) prq->type,
2331               GNUNET_h2s (key));
2332 #endif
2333   GNUNET_STATISTICS_update (stats,
2334                             gettext_noop ("# replies received and matched"),
2335                             1,
2336                             GNUNET_NO);
2337   if (prq->sender != NULL)
2338     {
2339       /* FIXME: should we be more precise here and not use
2340          "start_time" but a peer-specific time stamp? */
2341       cur_delay = GNUNET_TIME_absolute_get_duration (pr->start_time);
2342       prq->sender->avg_delay.value
2343         = (prq->sender->avg_delay.value *
2344            (RUNAVG_DELAY_N - 1) + cur_delay.value) / RUNAVG_DELAY_N;
2345       prq->sender->avg_priority
2346         = (prq->sender->avg_priority *
2347            (RUNAVG_DELAY_N - 1) + pr->priority) / (double) RUNAVG_DELAY_N;
2348       if (pr->cp != NULL)
2349         {
2350           GNUNET_PEER_change_rc (prq->sender->last_p2p_replies
2351                                  [prq->sender->last_p2p_replies_woff % P2P_SUCCESS_LIST_SIZE],
2352                                  -1);
2353           GNUNET_PEER_change_rc (pr->cp->pid, 1);
2354           prq->sender->last_p2p_replies
2355             [(prq->sender->last_p2p_replies_woff++) % P2P_SUCCESS_LIST_SIZE]
2356             = pr->cp->pid;
2357         }
2358       else
2359         {
2360           if (NULL != prq->sender->last_client_replies
2361               [(prq->sender->last_client_replies_woff) % CS2P_SUCCESS_LIST_SIZE])
2362             GNUNET_SERVER_client_drop (prq->sender->last_client_replies
2363                                        [(prq->sender->last_client_replies_woff) % CS2P_SUCCESS_LIST_SIZE]);
2364           prq->sender->last_client_replies
2365             [(prq->sender->last_client_replies_woff++) % CS2P_SUCCESS_LIST_SIZE]
2366             = pr->client_request_list->client_list->client;
2367           GNUNET_SERVER_client_keep (pr->client_request_list->client_list->client);
2368         }
2369     }
2370   GNUNET_CRYPTO_hash (prq->data,
2371                       prq->size,
2372                       &chash);
2373   switch (prq->type)
2374     {
2375     case GNUNET_BLOCK_TYPE_DBLOCK:
2376     case GNUNET_BLOCK_TYPE_IBLOCK:
2377       /* only possible reply, stop requesting! */
2378       while (NULL != pr->pending_head)
2379         destroy_pending_message_list_entry (pr->pending_head);
2380       if (pr->qe != NULL)
2381         {
2382           if (pr->client_request_list != NULL)
2383             GNUNET_SERVER_receive_done (pr->client_request_list->client_list->client,
2384                                         GNUNET_YES);
2385           GNUNET_DATASTORE_cancel (pr->qe);
2386           pr->qe = NULL;
2387         }
2388       pr->do_remove = GNUNET_YES;
2389       if (pr->task != GNUNET_SCHEDULER_NO_TASK)
2390         {
2391           GNUNET_SCHEDULER_cancel (sched,
2392                                    pr->task);
2393           pr->task = GNUNET_SCHEDULER_NO_TASK;
2394         }
2395       GNUNET_break (GNUNET_YES ==
2396                     GNUNET_CONTAINER_multihashmap_remove (query_request_map,
2397                                                           key,
2398                                                           pr));
2399       break;
2400     case GNUNET_BLOCK_TYPE_SBLOCK:
2401       if (pr->namespace == NULL)
2402         {
2403           GNUNET_break (0);
2404           return GNUNET_YES;
2405         }
2406       if (0 != memcmp (pr->namespace,
2407                        &prq->namespace,
2408                        sizeof (GNUNET_HashCode)))
2409         {
2410           GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
2411                       _("Reply mismatched in terms of namespace.  Discarded.\n"));
2412           return GNUNET_YES; /* wrong namespace */
2413         }
2414       /* then: fall-through! */
2415     case GNUNET_BLOCK_TYPE_KBLOCK:
2416     case GNUNET_BLOCK_TYPE_NBLOCK:
2417       if (pr->bf != NULL)
2418         {
2419           mingle_hash (&chash, pr->mingle, &mhash);
2420           if (GNUNET_YES == GNUNET_CONTAINER_bloomfilter_test (pr->bf,
2421                                                                &mhash))
2422             {
2423               GNUNET_STATISTICS_update (stats,
2424                                         gettext_noop ("# duplicate replies discarded (bloomfilter)"),
2425                                         1,
2426                                         GNUNET_NO);
2427 #if DEBUG_FS
2428               GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2429                           "Duplicate response `%s', discarding.\n",
2430                           GNUNET_h2s (&mhash));
2431 #endif
2432               return GNUNET_YES; /* duplicate */
2433             }
2434 #if DEBUG_FS
2435           GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2436                       "New response `%s', adding to filter.\n",
2437                       GNUNET_h2s (&mhash));
2438 #endif
2439         }
2440       if (pr->client_request_list != NULL)
2441         {
2442           if (pr->replies_seen_size == pr->replies_seen_off)
2443             GNUNET_array_grow (pr->replies_seen,
2444                                pr->replies_seen_size,
2445                                pr->replies_seen_size * 2 + 4);
2446             pr->replies_seen[pr->replies_seen_off++] = chash;
2447         }
2448       if ( (pr->bf == NULL) ||
2449            (pr->client_request_list != NULL) )
2450         refresh_bloomfilter (pr);
2451       GNUNET_CONTAINER_bloomfilter_add (pr->bf,
2452                                         &mhash);
2453       break;
2454     default:
2455       GNUNET_break (0);
2456       return GNUNET_YES;
2457     }
2458   prq->priority += pr->remaining_priority;
2459   pr->remaining_priority = 0;
2460   if (NULL != pr->client_request_list)
2461     {
2462       GNUNET_STATISTICS_update (stats,
2463                                 gettext_noop ("# replies received for local clients"),
2464                                 1,
2465                                 GNUNET_NO);
2466       cl = pr->client_request_list->client_list;
2467       msize = sizeof (struct PutMessage) + prq->size;
2468       creply = GNUNET_malloc (msize + sizeof (struct ClientResponseMessage));
2469       creply->msize = msize;
2470       creply->client_list = cl;
2471       GNUNET_CONTAINER_DLL_insert_after (cl->res_head,
2472                                          cl->res_tail,
2473                                          cl->res_tail,
2474                                          creply);
2475       pm = (struct PutMessage*) &creply[1];
2476       pm->header.type = htons (GNUNET_MESSAGE_TYPE_FS_PUT);
2477       pm->header.size = htons (msize);
2478       pm->type = htonl (prq->type);
2479       pm->expiration = GNUNET_TIME_absolute_hton (prq->expiration);
2480       memcpy (&pm[1], prq->data, prq->size);
2481       if (NULL == cl->th)
2482         {
2483 #if DEBUG_FS
2484           GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2485                       "Transmitting result for query `%s' to client\n",
2486                       GNUNET_h2s (key));
2487 #endif
2488           cl->th = GNUNET_SERVER_notify_transmit_ready (cl->client,
2489                                                         msize,
2490                                                         GNUNET_TIME_UNIT_FOREVER_REL,
2491                                                         &transmit_to_client,
2492                                                         cl);
2493         }
2494       GNUNET_break (cl->th != NULL);
2495       if (pr->do_remove)
2496         {
2497           prq->finished = GNUNET_YES;
2498           destroy_pending_request (pr);
2499         }
2500     }
2501   else
2502     {
2503       cp = pr->cp;
2504 #if DEBUG_FS
2505       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2506                   "Transmitting result for query `%s' to other peer (PID=%u)\n",
2507                   GNUNET_h2s (key),
2508                   (unsigned int) cp->pid);
2509 #endif
2510       GNUNET_STATISTICS_update (stats,
2511                                 gettext_noop ("# replies received for other peers"),
2512                                 1,
2513                                 GNUNET_NO);
2514       msize = sizeof (struct PutMessage) + prq->size;
2515       reply = GNUNET_malloc (msize + sizeof (struct PendingMessage));
2516       reply->cont = &transmit_reply_continuation;
2517       reply->cont_cls = pr;
2518       reply->msize = msize;
2519       reply->priority = UINT32_MAX; /* send replies first! */
2520       pm = (struct PutMessage*) &reply[1];
2521       pm->header.type = htons (GNUNET_MESSAGE_TYPE_FS_PUT);
2522       pm->header.size = htons (msize);
2523       pm->type = htonl (prq->type);
2524       pm->expiration = GNUNET_TIME_absolute_hton (prq->expiration);
2525       memcpy (&pm[1], prq->data, prq->size);
2526       add_to_pending_messages_for_peer (cp, reply, pr);
2527     }
2528   return GNUNET_YES;
2529 }
2530
2531
2532 /**
2533  * Continuation called to notify client about result of the
2534  * operation.
2535  *
2536  * @param cls closure
2537  * @param success GNUNET_SYSERR on failure
2538  * @param msg NULL on success, otherwise an error message
2539  */
2540 static void
2541 put_migration_continuation (void *cls,
2542                             int success,
2543                             const char *msg)
2544 {
2545   /* FIXME */
2546 }
2547
2548
2549 /**
2550  * Handle P2P "PUT" message.
2551  *
2552  * @param cls closure, always NULL
2553  * @param other the other peer involved (sender or receiver, NULL
2554  *        for loopback messages where we are both sender and receiver)
2555  * @param message the actual message
2556  * @param latency reported latency of the connection with 'other'
2557  * @param distance reported distance (DV) to 'other'
2558  * @return GNUNET_OK to keep the connection open,
2559  *         GNUNET_SYSERR to close it (signal serious error)
2560  */
2561 static int
2562 handle_p2p_put (void *cls,
2563                 const struct GNUNET_PeerIdentity *other,
2564                 const struct GNUNET_MessageHeader *message,
2565                 struct GNUNET_TIME_Relative latency,
2566                 uint32_t distance)
2567 {
2568   const struct PutMessage *put;
2569   uint16_t msize;
2570   size_t dsize;
2571   enum GNUNET_BLOCK_Type type;
2572   struct GNUNET_TIME_Absolute expiration;
2573   GNUNET_HashCode query;
2574   struct ProcessReplyClosure prq;
2575   const struct SBlock *sb;
2576
2577   msize = ntohs (message->size);
2578   if (msize < sizeof (struct PutMessage))
2579     {
2580       GNUNET_break_op(0);
2581       return GNUNET_SYSERR;
2582     }
2583   put = (const struct PutMessage*) message;
2584   dsize = msize - sizeof (struct PutMessage);
2585   type = ntohl (put->type);
2586   expiration = GNUNET_TIME_absolute_ntoh (put->expiration);
2587
2588   if (GNUNET_OK !=
2589       GNUNET_BLOCK_check_block (type,
2590                                 &put[1],
2591                                 dsize,
2592                                 &query))
2593     {
2594       GNUNET_break_op (0);
2595       return GNUNET_SYSERR;
2596     }
2597   if (type == GNUNET_BLOCK_TYPE_ONDEMAND)
2598     return GNUNET_SYSERR;
2599   if (GNUNET_BLOCK_TYPE_SBLOCK == type)
2600     {
2601       sb = (const struct SBlock*) &put[1];
2602       GNUNET_CRYPTO_hash (&sb->subspace,
2603                           sizeof (struct GNUNET_CRYPTO_RsaPublicKeyBinaryEncoded),
2604                           &prq.namespace);
2605     }
2606
2607 #if DEBUG_FS
2608   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2609               "Received result for query `%s' from peer `%4s'\n",
2610               GNUNET_h2s (&query),
2611               GNUNET_i2s (other));
2612 #endif
2613   GNUNET_STATISTICS_update (stats,
2614                             gettext_noop ("# replies received (overall)"),
2615                             1,
2616                             GNUNET_NO);
2617   /* now, lookup 'query' */
2618   prq.data = (const void*) &put[1];
2619   if (other != NULL)
2620     prq.sender = GNUNET_CONTAINER_multihashmap_get (connected_peers,
2621                                                     &other->hashPubKey);
2622   prq.size = dsize;
2623   prq.type = type;
2624   prq.expiration = expiration;
2625   prq.priority = 0;
2626   prq.finished = GNUNET_NO;
2627   GNUNET_CONTAINER_multihashmap_get_multiple (query_request_map,
2628                                               &query,
2629                                               &process_reply,
2630                                               &prq);
2631   if (GNUNET_YES == active_migration)
2632     {
2633 #if DEBUG_FS
2634       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2635                   "Replicating result for query `%s' with priority %u\n",
2636                   GNUNET_h2s (&query),
2637                   prq.priority);
2638 #endif
2639       GNUNET_DATASTORE_put (dsh,
2640                             0, &query, dsize, &put[1],
2641                             type, prq.priority, 1 /* anonymity */,
2642                             expiration,
2643                             1 + prq.priority, MAX_DATASTORE_QUEUE,
2644                             GNUNET_CONSTANTS_SERVICE_TIMEOUT,
2645                             &put_migration_continuation,
2646                             NULL);
2647     }
2648   return GNUNET_OK;
2649 }
2650
2651
2652 /* **************************** P2P GET Handling ************************ */
2653
2654
2655 /**
2656  * Closure for 'check_duplicate_request_{peer,client}'.
2657  */
2658 struct CheckDuplicateRequestClosure
2659 {
2660   /**
2661    * The new request we should check if it already exists.
2662    */
2663   const struct PendingRequest *pr;
2664
2665   /**
2666    * Existing request found by the checker, NULL if none.
2667    */
2668   struct PendingRequest *have;
2669 };
2670
2671
2672 /**
2673  * Iterator over entries in the 'query_request_map' that
2674  * tries to see if we have the same request pending from
2675  * the same client already.
2676  *
2677  * @param cls closure (our 'struct CheckDuplicateRequestClosure')
2678  * @param key current key code (query, ignored, must match)
2679  * @param value value in the hash map (a 'struct PendingRequest'
2680  *              that already exists)
2681  * @return GNUNET_YES if we should continue to
2682  *         iterate (no match yet)
2683  *         GNUNET_NO if not (match found).
2684  */
2685 static int
2686 check_duplicate_request_client (void *cls,
2687                                 const GNUNET_HashCode * key,
2688                                 void *value)
2689 {
2690   struct CheckDuplicateRequestClosure *cdc = cls;
2691   struct PendingRequest *have = value;
2692
2693   if (have->client_request_list == NULL)
2694     return GNUNET_YES;
2695   if ( (cdc->pr->client_request_list->client_list->client == have->client_request_list->client_list->client) &&
2696        (cdc->pr != have) )
2697     {
2698       cdc->have = have;
2699       return GNUNET_NO;
2700     }
2701   return GNUNET_YES;
2702 }
2703
2704
2705 /**
2706  * We're processing (local) results for a search request
2707  * from another peer.  Pass applicable results to the
2708  * peer and if we are done either clean up (operation
2709  * complete) or forward to other peers (more results possible).
2710  *
2711  * @param cls our closure (struct LocalGetContext)
2712  * @param key key for the content
2713  * @param size number of bytes in data
2714  * @param data content stored
2715  * @param type type of the content
2716  * @param priority priority of the content
2717  * @param anonymity anonymity-level for the content
2718  * @param expiration expiration time for the content
2719  * @param uid unique identifier for the datum;
2720  *        maybe 0 if no unique identifier is available
2721  */
2722 static void
2723 process_local_reply (void *cls,
2724                      const GNUNET_HashCode * key,
2725                      uint32_t size,
2726                      const void *data,
2727                      enum GNUNET_BLOCK_Type type,
2728                      uint32_t priority,
2729                      uint32_t anonymity,
2730                      struct GNUNET_TIME_Absolute
2731                      expiration,
2732                      uint64_t uid)
2733 {
2734   struct PendingRequest *pr = cls;
2735   struct ProcessReplyClosure prq;
2736   struct CheckDuplicateRequestClosure cdrc;
2737   const struct SBlock *sb;
2738   GNUNET_HashCode dhash;
2739   GNUNET_HashCode mhash;
2740   GNUNET_HashCode query;
2741
2742   if (NULL == key)
2743     {
2744 #if DEBUG_FS > 1
2745       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2746                   "Done processing local replies, forwarding request to other peers.\n");
2747 #endif
2748       pr->qe = NULL;
2749       if (pr->client_request_list != NULL)
2750         {
2751           GNUNET_SERVER_receive_done (pr->client_request_list->client_list->client,
2752                                       GNUNET_YES);
2753           /* Figure out if this is a duplicate request and possibly
2754              merge 'struct PendingRequest' entries */
2755           cdrc.have = NULL;
2756           cdrc.pr = pr;
2757           GNUNET_CONTAINER_multihashmap_get_multiple (query_request_map,
2758                                                       &pr->query,
2759                                                       &check_duplicate_request_client,
2760                                                       &cdrc);
2761           if (cdrc.have != NULL)
2762             {
2763 #if DEBUG_FS
2764               GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2765                           "Received request for block `%s' twice from client, will only request once.\n",
2766                           GNUNET_h2s (&pr->query));
2767 #endif
2768
2769               destroy_pending_request (pr);
2770               return;
2771             }
2772         }
2773
2774       /* no more results */
2775       if (pr->task == GNUNET_SCHEDULER_NO_TASK)
2776         pr->task = GNUNET_SCHEDULER_add_now (sched,
2777                                              &forward_request_task,
2778                                              pr);
2779       return;
2780     }
2781 #if DEBUG_FS
2782   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2783               "New local response to `%s' of type %u.\n",
2784               GNUNET_h2s (key),
2785               type);
2786 #endif
2787   if (type == GNUNET_BLOCK_TYPE_ONDEMAND)
2788     {
2789 #if DEBUG_FS
2790       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2791                   "Found ONDEMAND block, performing on-demand encoding\n");
2792 #endif
2793       GNUNET_STATISTICS_update (stats,
2794                                 gettext_noop ("# on-demand blocks matched requests"),
2795                                 1,
2796                                 GNUNET_NO);
2797       if (GNUNET_OK !=
2798           GNUNET_FS_handle_on_demand_block (key, size, data, type, priority,
2799                                             anonymity, expiration, uid,
2800                                             &process_local_reply,
2801                                             pr))
2802       if (pr->qe != NULL)
2803         GNUNET_DATASTORE_get_next (dsh, GNUNET_YES);
2804       return;
2805     }
2806   /* check for duplicates */
2807   GNUNET_CRYPTO_hash (data, size, &dhash);
2808   mingle_hash (&dhash,
2809                pr->mingle,
2810                &mhash);
2811   if ( (pr->bf != NULL) &&
2812        (GNUNET_YES ==
2813         GNUNET_CONTAINER_bloomfilter_test (pr->bf,
2814                                            &mhash)) )
2815     {
2816 #if DEBUG_FS
2817       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2818                   "Result from datastore filtered by bloomfilter (duplicate).\n");
2819 #endif
2820       GNUNET_STATISTICS_update (stats,
2821                                 gettext_noop ("# results filtered by query bloomfilter"),
2822                                 1,
2823                                 GNUNET_NO);
2824       if (pr->qe != NULL)
2825         GNUNET_DATASTORE_get_next (dsh, GNUNET_YES);
2826       return;
2827     }
2828 #if DEBUG_FS
2829   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2830               "Found result for query `%s' in local datastore\n",
2831               GNUNET_h2s (key));
2832 #endif
2833   GNUNET_STATISTICS_update (stats,
2834                             gettext_noop ("# results found locally"),
2835                             1,
2836                             GNUNET_NO);
2837   pr->results_found++;
2838   memset (&prq, 0, sizeof (prq));
2839   prq.data = data;
2840   prq.expiration = expiration;
2841   prq.size = size;
2842   if (GNUNET_BLOCK_TYPE_SBLOCK == type)
2843     {
2844       sb = (const struct SBlock*) data;
2845       GNUNET_CRYPTO_hash (&sb->subspace,
2846                           sizeof (struct GNUNET_CRYPTO_RsaPublicKeyBinaryEncoded),
2847                           &prq.namespace);
2848     }
2849   if (GNUNET_OK != GNUNET_BLOCK_check_block (type,
2850                                              data,
2851                                              size,
2852                                              &query))
2853     {
2854       GNUNET_break (0);
2855       GNUNET_DATASTORE_remove (dsh,
2856                                key,
2857                                size, data,
2858                                -1, -1,
2859                                GNUNET_TIME_UNIT_FOREVER_REL,
2860                                NULL, NULL);
2861       GNUNET_DATASTORE_get_next (dsh, GNUNET_YES);
2862       return;
2863     }
2864   prq.type = type;
2865   prq.priority = priority;
2866   prq.finished = GNUNET_NO;
2867   process_reply (&prq, key, pr);
2868   if (prq.finished == GNUNET_YES)
2869     return;
2870   if (pr->qe == NULL)
2871     return; /* done here */
2872   if ( (type == GNUNET_BLOCK_TYPE_DBLOCK) ||
2873        (type == GNUNET_BLOCK_TYPE_IBLOCK) )
2874     {
2875       GNUNET_DATASTORE_get_next (dsh, GNUNET_NO);
2876       return;
2877     }
2878   if ( (pr->client_request_list == NULL) &&
2879        ( (GNUNET_YES == test_load_too_high()) ||
2880          (pr->results_found > 5 + 2 * pr->priority) ) )
2881     {
2882 #if DEBUG_FS > 2
2883       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2884                   "Load too high, done with request\n");
2885 #endif
2886       GNUNET_STATISTICS_update (stats,
2887                                 gettext_noop ("# processing result set cut short due to load"),
2888                                 1,
2889                                 GNUNET_NO);
2890       GNUNET_DATASTORE_get_next (dsh, GNUNET_NO);
2891       return;
2892     }
2893   GNUNET_DATASTORE_get_next (dsh, GNUNET_YES);
2894 }
2895
2896
2897 /**
2898  * We've received a request with the specified priority.  Bound it
2899  * according to how much we trust the given peer.
2900  *
2901  * @param prio_in requested priority
2902  * @param cp the peer making the request
2903  * @return effective priority
2904  */
2905 static uint32_t
2906 bound_priority (uint32_t prio_in,
2907                 struct ConnectedPeer *cp)
2908 {
2909   return 0; // FIXME!
2910 }
2911
2912
2913 /**
2914  * Iterator over entries in the 'query_request_map' that
2915  * tries to see if we have the same request pending from
2916  * the same peer already.
2917  *
2918  * @param cls closure (our 'struct CheckDuplicateRequestClosure')
2919  * @param key current key code (query, ignored, must match)
2920  * @param value value in the hash map (a 'struct PendingRequest'
2921  *              that already exists)
2922  * @return GNUNET_YES if we should continue to
2923  *         iterate (no match yet)
2924  *         GNUNET_NO if not (match found).
2925  */
2926 static int
2927 check_duplicate_request_peer (void *cls,
2928                               const GNUNET_HashCode * key,
2929                               void *value)
2930 {
2931   struct CheckDuplicateRequestClosure *cdc = cls;
2932   struct PendingRequest *have = value;
2933
2934   if (cdc->pr->target_pid == have->target_pid)
2935     {
2936       cdc->have = have;
2937       return GNUNET_NO;
2938     }
2939   return GNUNET_YES;
2940 }
2941
2942
2943 /**
2944  * Handle P2P "GET" request.
2945  *
2946  * @param cls closure, always NULL
2947  * @param other the other peer involved (sender or receiver, NULL
2948  *        for loopback messages where we are both sender and receiver)
2949  * @param message the actual message
2950  * @param latency reported latency of the connection with 'other'
2951  * @param distance reported distance (DV) to 'other'
2952  * @return GNUNET_OK to keep the connection open,
2953  *         GNUNET_SYSERR to close it (signal serious error)
2954  */
2955 static int
2956 handle_p2p_get (void *cls,
2957                 const struct GNUNET_PeerIdentity *other,
2958                 const struct GNUNET_MessageHeader *message,
2959                 struct GNUNET_TIME_Relative latency,
2960                 uint32_t distance)
2961 {
2962   struct PendingRequest *pr;
2963   struct ConnectedPeer *cp;
2964   struct ConnectedPeer *cps;
2965   struct CheckDuplicateRequestClosure cdc;
2966   struct GNUNET_TIME_Relative timeout;
2967   uint16_t msize;
2968   const struct GetMessage *gm;
2969   unsigned int bits;
2970   const GNUNET_HashCode *opt;
2971   uint32_t bm;
2972   size_t bfsize;
2973   uint32_t ttl_decrement;
2974   enum GNUNET_BLOCK_Type type;
2975   double preference;
2976   int have_ns;
2977
2978   msize = ntohs(message->size);
2979   if (msize < sizeof (struct GetMessage))
2980     {
2981       GNUNET_break_op (0);
2982       return GNUNET_SYSERR;
2983     }
2984   gm = (const struct GetMessage*) message;
2985   type = ntohl (gm->type);
2986   switch (type)
2987     {
2988     case GNUNET_BLOCK_TYPE_ANY:
2989     case GNUNET_BLOCK_TYPE_DBLOCK:
2990     case GNUNET_BLOCK_TYPE_IBLOCK:
2991     case GNUNET_BLOCK_TYPE_KBLOCK:
2992     case GNUNET_BLOCK_TYPE_SBLOCK:
2993       break;
2994     default:
2995       GNUNET_break_op (0);
2996       return GNUNET_SYSERR;
2997     }
2998   bm = ntohl (gm->hash_bitmap);
2999   bits = 0;
3000   while (bm > 0)
3001     {
3002       if (1 == (bm & 1))
3003         bits++;
3004       bm >>= 1;
3005     }
3006   if (msize < sizeof (struct GetMessage) + bits * sizeof (GNUNET_HashCode))
3007     {
3008       GNUNET_break_op (0);
3009       return GNUNET_SYSERR;
3010     }
3011   opt = (const GNUNET_HashCode*) &gm[1];
3012   bfsize = msize - sizeof (struct GetMessage) + bits * sizeof (GNUNET_HashCode);
3013   bm = ntohl (gm->hash_bitmap);
3014   if ( (0 != (bm & GET_MESSAGE_BIT_SKS_NAMESPACE)) &&
3015        (type != GNUNET_BLOCK_TYPE_SBLOCK) )
3016     {
3017       GNUNET_break_op (0);
3018       return GNUNET_SYSERR;
3019     }
3020   bits = 0;
3021   cps = GNUNET_CONTAINER_multihashmap_get (connected_peers,
3022                                            &other->hashPubKey);
3023   if (NULL == cps)
3024     {
3025       /* peer must have just disconnected */
3026       GNUNET_STATISTICS_update (stats,
3027                                 gettext_noop ("# requests dropped due to initiator not being connected"),
3028                                 1,
3029                                 GNUNET_NO);
3030       return GNUNET_SYSERR;
3031     }
3032   if (0 != (bm & GET_MESSAGE_BIT_RETURN_TO))
3033     cp = GNUNET_CONTAINER_multihashmap_get (connected_peers,
3034                                             &opt[bits++]);
3035   else
3036     cp = cps;
3037   if (cp == NULL)
3038     {
3039 #if DEBUG_FS
3040       if (0 != (bm & GET_MESSAGE_BIT_RETURN_TO))
3041         GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
3042                     "Failed to find RETURN-TO peer `%4s' in connection set. Dropping query.\n",
3043                     GNUNET_i2s ((const struct GNUNET_PeerIdentity*) &opt[bits-1]));
3044
3045       else
3046         GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
3047                     "Failed to find peer `%4s' in connection set. Dropping query.\n",
3048                     GNUNET_i2s (other));
3049 #endif
3050       GNUNET_STATISTICS_update (stats,
3051                                 gettext_noop ("# requests dropped due to missing reverse route"),
3052                                 1,
3053                                 GNUNET_NO);
3054      /* FIXME: try connect? */
3055       return GNUNET_OK;
3056     }
3057   /* note that we can really only check load here since otherwise
3058      peers could find out that we are overloaded by not being
3059      disconnected after sending us a malformed query... */
3060   if (GNUNET_YES == test_load_too_high ())
3061     {
3062 #if DEBUG_FS
3063       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
3064                   "Dropping query from `%s', this peer is too busy.\n",
3065                   GNUNET_i2s (other));
3066 #endif
3067       GNUNET_STATISTICS_update (stats,
3068                                 gettext_noop ("# requests dropped due to high load"),
3069                                 1,
3070                                 GNUNET_NO);
3071       return GNUNET_OK;
3072     }
3073
3074 #if DEBUG_FS
3075   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
3076               "Received request for `%s' of type %u from peer `%4s' with flags %u\n",
3077               GNUNET_h2s (&gm->query),
3078               (unsigned int) type,
3079               GNUNET_i2s (other),
3080               (unsigned int) bm);
3081 #endif
3082   have_ns = (0 != (bm & GET_MESSAGE_BIT_SKS_NAMESPACE));
3083   pr = GNUNET_malloc (sizeof (struct PendingRequest) +
3084                       (have_ns ? sizeof(GNUNET_HashCode) : 0));
3085   if (have_ns)
3086     pr->namespace = (GNUNET_HashCode*) &pr[1];
3087   pr->type = type;
3088   pr->mingle = ntohl (gm->filter_mutator);
3089   if (0 != (bm & GET_MESSAGE_BIT_SKS_NAMESPACE))
3090     memcpy (&pr[1], &opt[bits++], sizeof (GNUNET_HashCode));
3091   if (0 != (bm & GET_MESSAGE_BIT_TRANSMIT_TO))
3092     pr->target_pid = GNUNET_PEER_intern ((const struct GNUNET_PeerIdentity*) &opt[bits++]);
3093
3094   pr->anonymity_level = 1;
3095   pr->priority = bound_priority (ntohl (gm->priority), cps);
3096   pr->ttl = bound_ttl (ntohl (gm->ttl), pr->priority);
3097   pr->query = gm->query;
3098   /* decrement ttl (always) */
3099   ttl_decrement = 2 * TTL_DECREMENT +
3100     GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
3101                               TTL_DECREMENT);
3102   if ( (pr->ttl < 0) &&
3103        (((int32_t)(pr->ttl - ttl_decrement)) > 0) )
3104     {
3105 #if DEBUG_FS
3106       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
3107                   "Dropping query from `%s' due to TTL underflow (%d - %u).\n",
3108                   GNUNET_i2s (other),
3109                   pr->ttl,
3110                   ttl_decrement);
3111 #endif
3112       GNUNET_STATISTICS_update (stats,
3113                                 gettext_noop ("# requests dropped due TTL underflow"),
3114                                 1,
3115                                 GNUNET_NO);
3116       /* integer underflow => drop (should be very rare)! */
3117       GNUNET_free (pr);
3118       return GNUNET_OK;
3119     }
3120   pr->ttl -= ttl_decrement;
3121   pr->start_time = GNUNET_TIME_absolute_get ();
3122
3123   /* get bloom filter */
3124   if (bfsize > 0)
3125     {
3126       pr->bf = GNUNET_CONTAINER_bloomfilter_init ((const char*) &opt[bits],
3127                                                   bfsize,
3128                                                   BLOOMFILTER_K);
3129       pr->bf_size = bfsize;
3130     }
3131
3132   cdc.have = NULL;
3133   cdc.pr = pr;
3134   GNUNET_CONTAINER_multihashmap_get_multiple (query_request_map,
3135                                               &gm->query,
3136                                               &check_duplicate_request_peer,
3137                                               &cdc);
3138   if (cdc.have != NULL)
3139     {
3140       if (cdc.have->start_time.value + cdc.have->ttl >=
3141           pr->start_time.value + pr->ttl)
3142         {
3143           /* existing request has higher TTL, drop new one! */
3144           cdc.have->priority += pr->priority;
3145           destroy_pending_request (pr);
3146 #if DEBUG_FS
3147           GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
3148                       "Have existing request with higher TTL, dropping new request.\n",
3149                       GNUNET_i2s (other));
3150 #endif
3151           GNUNET_STATISTICS_update (stats,
3152                                     gettext_noop ("# requests dropped due to higher-TTL request"),
3153                                     1,
3154                                     GNUNET_NO);
3155           return GNUNET_OK;
3156         }
3157       else
3158         {
3159           /* existing request has lower TTL, drop old one! */
3160           pr->priority += cdc.have->priority;
3161           /* Possible optimization: if we have applicable pending
3162              replies in 'cdc.have', we might want to move those over
3163              (this is a really rare special-case, so it is not clear
3164              that this would be worth it) */
3165           destroy_pending_request (cdc.have);
3166           /* keep processing 'pr'! */
3167         }
3168     }
3169
3170   pr->cp = cp;
3171   GNUNET_break (GNUNET_OK ==
3172                 GNUNET_CONTAINER_multihashmap_put (query_request_map,
3173                                                    &gm->query,
3174                                                    pr,
3175                                                    GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE));
3176   GNUNET_break (GNUNET_OK ==
3177                 GNUNET_CONTAINER_multihashmap_put (peer_request_map,
3178                                                    &other->hashPubKey,
3179                                                    pr,
3180                                                    GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE));
3181
3182   pr->hnode = GNUNET_CONTAINER_heap_insert (requests_by_expiration_heap,
3183                                             pr,
3184                                             pr->start_time.value + pr->ttl);
3185
3186   GNUNET_STATISTICS_update (stats,
3187                             gettext_noop ("# P2P searches received"),
3188                             1,
3189                             GNUNET_NO);
3190   GNUNET_STATISTICS_update (stats,
3191                             gettext_noop ("# P2P searches active"),
3192                             1,
3193                             GNUNET_NO);
3194
3195   /* calculate change in traffic preference */
3196   preference = (double) pr->priority;
3197   if (preference < QUERY_BANDWIDTH_VALUE)
3198     preference = QUERY_BANDWIDTH_VALUE;
3199   cps->inc_preference += preference;
3200
3201   /* process locally */
3202   if (type == GNUNET_BLOCK_TYPE_DBLOCK)
3203     type = GNUNET_BLOCK_TYPE_ANY; /* to get on-demand as well */
3204   timeout = GNUNET_TIME_relative_multiply (BASIC_DATASTORE_REQUEST_DELAY,
3205                                            (pr->priority + 1));
3206   pr->qe = GNUNET_DATASTORE_get (dsh,
3207                                  &gm->query,
3208                                  type,
3209                                  pr->priority + 1,
3210                                  MAX_DATASTORE_QUEUE,
3211                                  timeout,
3212                                  &process_local_reply,
3213                                  pr);
3214
3215   /* Are multiple results possible?  If so, start processing remotely now! */
3216   switch (pr->type)
3217     {
3218     case GNUNET_BLOCK_TYPE_DBLOCK:
3219     case GNUNET_BLOCK_TYPE_IBLOCK:
3220       /* only one result, wait for datastore */
3221       break;
3222     default:
3223       if (pr->task == GNUNET_SCHEDULER_NO_TASK)
3224         pr->task = GNUNET_SCHEDULER_add_now (sched,
3225                                              &forward_request_task,
3226                                              pr);
3227     }
3228
3229   /* make sure we don't track too many requests */
3230   if (GNUNET_CONTAINER_heap_get_size (requests_by_expiration_heap) > max_pending_requests)
3231     {
3232       pr = GNUNET_CONTAINER_heap_peek (requests_by_expiration_heap);
3233       destroy_pending_request (pr);
3234     }
3235   return GNUNET_OK;
3236 }
3237
3238
3239 /* **************************** CS GET Handling ************************ */
3240
3241
3242 /**
3243  * Handle START_SEARCH-message (search request from client).
3244  *
3245  * @param cls closure
3246  * @param client identification of the client
3247  * @param message the actual message
3248  */
3249 static void
3250 handle_start_search (void *cls,
3251                      struct GNUNET_SERVER_Client *client,
3252                      const struct GNUNET_MessageHeader *message)
3253 {
3254   static GNUNET_HashCode all_zeros;
3255   const struct SearchMessage *sm;
3256   struct ClientList *cl;
3257   struct ClientRequestList *crl;
3258   struct PendingRequest *pr;
3259   uint16_t msize;
3260   unsigned int sc;
3261   enum GNUNET_BLOCK_Type type;
3262
3263   msize = ntohs (message->size);
3264   if ( (msize < sizeof (struct SearchMessage)) ||
3265        (0 != (msize - sizeof (struct SearchMessage)) % sizeof (GNUNET_HashCode)) )
3266     {
3267       GNUNET_break (0);
3268       GNUNET_SERVER_receive_done (client,
3269                                   GNUNET_SYSERR);
3270       return;
3271     }
3272   GNUNET_STATISTICS_update (stats,
3273                             gettext_noop ("# client searches received"),
3274                             1,
3275                             GNUNET_NO);
3276   sc = (msize - sizeof (struct SearchMessage)) / sizeof (GNUNET_HashCode);
3277   sm = (const struct SearchMessage*) message;
3278   type = ntohl (sm->type);
3279 #if DEBUG_FS
3280   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
3281               "Received request for `%s' of type %u from local client\n",
3282               GNUNET_h2s (&sm->query),
3283               (unsigned int) type);
3284 #endif
3285   switch (type)
3286     {
3287     case GNUNET_BLOCK_TYPE_ANY:
3288     case GNUNET_BLOCK_TYPE_DBLOCK:
3289     case GNUNET_BLOCK_TYPE_IBLOCK:
3290     case GNUNET_BLOCK_TYPE_KBLOCK:
3291     case GNUNET_BLOCK_TYPE_SBLOCK:
3292     case GNUNET_BLOCK_TYPE_NBLOCK:
3293       break;
3294     default:
3295       GNUNET_break (0);
3296       GNUNET_SERVER_receive_done (client,
3297                                   GNUNET_SYSERR);
3298       return;
3299     }
3300
3301   cl = client_list;
3302   while ( (cl != NULL) &&
3303           (cl->client != client) )
3304     cl = cl->next;
3305   if (cl == NULL)
3306     {
3307       cl = GNUNET_malloc (sizeof (struct ClientList));
3308       cl->client = client;
3309       GNUNET_SERVER_client_keep (client);
3310       cl->next = client_list;
3311       client_list = cl;
3312     }
3313   /* detect duplicate KBLOCK requests */
3314   if ( (type == GNUNET_BLOCK_TYPE_KBLOCK) ||
3315        (type == GNUNET_BLOCK_TYPE_NBLOCK) ||
3316        (type == GNUNET_BLOCK_TYPE_ANY) )
3317     {
3318       crl = cl->rl_head;
3319       while ( (crl != NULL) &&
3320               ( (0 != memcmp (&crl->req->query,
3321                               &sm->query,
3322                               sizeof (GNUNET_HashCode))) ||
3323                 (crl->req->type != type) ) )
3324         crl = crl->next;
3325       if (crl != NULL)
3326         {
3327 #if DEBUG_FS
3328           GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
3329                       "Have existing request, merging content-seen lists.\n");
3330 #endif
3331           pr = crl->req;
3332           /* Duplicate request (used to send long list of
3333              known/blocked results); merge 'pr->replies_seen'
3334              and update bloom filter */
3335           GNUNET_array_grow (pr->replies_seen,
3336                              pr->replies_seen_size,
3337                              pr->replies_seen_off + sc);
3338           memcpy (&pr->replies_seen[pr->replies_seen_off],
3339                   &sm[1],
3340                   sc * sizeof (GNUNET_HashCode));
3341           pr->replies_seen_off += sc;
3342           refresh_bloomfilter (pr);
3343           GNUNET_STATISTICS_update (stats,
3344                                     gettext_noop ("# client searches updated (merged content seen list)"),
3345                                     1,
3346                                     GNUNET_NO);
3347           GNUNET_SERVER_receive_done (client,
3348                                       GNUNET_OK);
3349           return;
3350         }
3351     }
3352   GNUNET_STATISTICS_update (stats,
3353                             gettext_noop ("# client searches active"),
3354                             1,
3355                             GNUNET_NO);
3356   pr = GNUNET_malloc (sizeof (struct PendingRequest) +
3357                       ((type == GNUNET_BLOCK_TYPE_SBLOCK) ? sizeof(GNUNET_HashCode) : 0));
3358   crl = GNUNET_malloc (sizeof (struct ClientRequestList));
3359   memset (crl, 0, sizeof (struct ClientRequestList));
3360   crl->client_list = cl;
3361   GNUNET_CONTAINER_DLL_insert (cl->rl_head,
3362                                cl->rl_tail,
3363                                crl);
3364   crl->req = pr;
3365   pr->type = type;
3366   pr->client_request_list = crl;
3367   GNUNET_array_grow (pr->replies_seen,
3368                      pr->replies_seen_size,
3369                      sc);
3370   memcpy (pr->replies_seen,
3371           &sm[1],
3372           sc * sizeof (GNUNET_HashCode));
3373   pr->replies_seen_off = sc;
3374   pr->anonymity_level = ntohl (sm->anonymity_level);
3375   refresh_bloomfilter (pr);
3376   pr->query = sm->query;
3377   if (0 == (1 & ntohl (sm->options)))
3378     pr->local_only = GNUNET_NO;
3379   else
3380     pr->local_only = GNUNET_YES;
3381   switch (type)
3382     {
3383     case GNUNET_BLOCK_TYPE_DBLOCK:
3384     case GNUNET_BLOCK_TYPE_IBLOCK:
3385       if (0 != memcmp (&sm->target,
3386                        &all_zeros,
3387                        sizeof (GNUNET_HashCode)))
3388         pr->target_pid = GNUNET_PEER_intern ((const struct GNUNET_PeerIdentity*) &sm->target);
3389       break;
3390     case GNUNET_BLOCK_TYPE_SBLOCK:
3391       pr->namespace = (GNUNET_HashCode*) &pr[1];
3392       memcpy (&pr[1], &sm->target, sizeof (GNUNET_HashCode));
3393       break;
3394     default:
3395       break;
3396     }
3397   GNUNET_break (GNUNET_OK ==
3398                 GNUNET_CONTAINER_multihashmap_put (query_request_map,
3399                                                    &sm->query,
3400                                                    pr,
3401                                                    GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE));
3402   if (type == GNUNET_BLOCK_TYPE_DBLOCK)
3403     type = GNUNET_BLOCK_TYPE_ANY; /* get on-demand blocks too! */
3404   pr->qe = GNUNET_DATASTORE_get (dsh,
3405                                  &sm->query,
3406                                  type,
3407                                  -3, -1,
3408                                  GNUNET_CONSTANTS_SERVICE_TIMEOUT,
3409                                  &process_local_reply,
3410                                  pr);
3411 }
3412
3413
3414 /* **************************** Startup ************************ */
3415
3416
3417 /**
3418  * List of handlers for P2P messages
3419  * that we care about.
3420  */
3421 static struct GNUNET_CORE_MessageHandler p2p_handlers[] =
3422   {
3423     { &handle_p2p_get,
3424       GNUNET_MESSAGE_TYPE_FS_GET, 0 },
3425     { &handle_p2p_put,
3426       GNUNET_MESSAGE_TYPE_FS_PUT, 0 },
3427     { NULL, 0, 0 }
3428   };
3429
3430
3431 /**
3432  * List of handlers for the messages understood by this
3433  * service.
3434  */
3435 static struct GNUNET_SERVER_MessageHandler handlers[] = {
3436   {&GNUNET_FS_handle_index_start, NULL,
3437    GNUNET_MESSAGE_TYPE_FS_INDEX_START, 0},
3438   {&GNUNET_FS_handle_index_list_get, NULL,
3439    GNUNET_MESSAGE_TYPE_FS_INDEX_LIST_GET, sizeof(struct GNUNET_MessageHeader) },
3440   {&GNUNET_FS_handle_unindex, NULL, GNUNET_MESSAGE_TYPE_FS_UNINDEX,
3441    sizeof (struct UnindexMessage) },
3442   {&handle_start_search, NULL, GNUNET_MESSAGE_TYPE_FS_START_SEARCH,
3443    0 },
3444   {NULL, NULL, 0, 0}
3445 };
3446
3447
3448 /**
3449  * Process fs requests.
3450  *
3451  * @param s scheduler to use
3452  * @param server the initialized server
3453  * @param c configuration to use
3454  */
3455 static int
3456 main_init (struct GNUNET_SCHEDULER_Handle *s,
3457            struct GNUNET_SERVER_Handle *server,
3458            const struct GNUNET_CONFIGURATION_Handle *c)
3459 {
3460   sched = s;
3461   cfg = c;
3462   stats = GNUNET_STATISTICS_create (sched, "fs", cfg);
3463   min_migration_delay = GNUNET_TIME_UNIT_SECONDS; // FIXME: get from config
3464   connected_peers = GNUNET_CONTAINER_multihashmap_create (128); // FIXME: get size from config
3465   query_request_map = GNUNET_CONTAINER_multihashmap_create (128); // FIXME: get size from config
3466   peer_request_map = GNUNET_CONTAINER_multihashmap_create (128); // FIXME: get size from config
3467   requests_by_expiration_heap = GNUNET_CONTAINER_heap_create (GNUNET_CONTAINER_HEAP_ORDER_MIN);
3468   core = GNUNET_CORE_connect (sched,
3469                               cfg,
3470                               GNUNET_TIME_UNIT_FOREVER_REL,
3471                               NULL,
3472                               NULL,
3473                               &peer_connect_handler,
3474                               &peer_disconnect_handler,
3475                               NULL, GNUNET_NO,
3476                               NULL, GNUNET_NO,
3477                               p2p_handlers);
3478   if (NULL == core)
3479     {
3480       GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
3481                   _("Failed to connect to `%s' service.\n"),
3482                   "core");
3483       GNUNET_CONTAINER_multihashmap_destroy (connected_peers);
3484       connected_peers = NULL;
3485       GNUNET_CONTAINER_multihashmap_destroy (query_request_map);
3486       query_request_map = NULL;
3487       GNUNET_CONTAINER_heap_destroy (requests_by_expiration_heap);
3488       requests_by_expiration_heap = NULL;
3489       GNUNET_CONTAINER_multihashmap_destroy (peer_request_map);
3490       peer_request_map = NULL;
3491       if (dsh != NULL)
3492         {
3493           GNUNET_DATASTORE_disconnect (dsh, GNUNET_NO);
3494           dsh = NULL;
3495         }
3496       return GNUNET_SYSERR;
3497     }
3498   /* FIXME: distinguish between sending and storing in options? */
3499   if (active_migration)
3500     {
3501       GNUNET_log (GNUNET_ERROR_TYPE_INFO,
3502                   _("Content migration is enabled, will start to gather data\n"));
3503       consider_migration_gathering ();
3504     }
3505   GNUNET_SERVER_disconnect_notify (server,
3506                                    &handle_client_disconnect,
3507                                    NULL);
3508   GNUNET_SERVER_add_handlers (server, handlers);
3509   GNUNET_SCHEDULER_add_delayed (sched,
3510                                 GNUNET_TIME_UNIT_FOREVER_REL,
3511                                 &shutdown_task,
3512                                 NULL);
3513   return GNUNET_OK;
3514 }
3515
3516
3517 /**
3518  * Process fs requests.
3519  *
3520  * @param cls closure
3521  * @param sched scheduler to use
3522  * @param server the initialized server
3523  * @param cfg configuration to use
3524  */
3525 static void
3526 run (void *cls,
3527      struct GNUNET_SCHEDULER_Handle *sched,
3528      struct GNUNET_SERVER_Handle *server,
3529      const struct GNUNET_CONFIGURATION_Handle *cfg)
3530 {
3531   active_migration = GNUNET_CONFIGURATION_get_value_yesno (cfg,
3532                                                            "FS",
3533                                                            "ACTIVEMIGRATION");
3534   dsh = GNUNET_DATASTORE_connect (cfg,
3535                                   sched);
3536   if (dsh == NULL)
3537     {
3538       GNUNET_SCHEDULER_shutdown (sched);
3539       return;
3540     }
3541   if ( (GNUNET_OK != GNUNET_FS_indexing_init (sched, cfg, dsh)) ||
3542        (GNUNET_OK != main_init (sched, server, cfg)) )
3543     {
3544       GNUNET_SCHEDULER_shutdown (sched);
3545       GNUNET_DATASTORE_disconnect (dsh, GNUNET_NO);
3546       dsh = NULL;
3547       return;
3548     }
3549 }
3550
3551
3552 /**
3553  * The main function for the fs service.
3554  *
3555  * @param argc number of arguments from the command line
3556  * @param argv command line arguments
3557  * @return 0 ok, 1 on error
3558  */
3559 int
3560 main (int argc, char *const *argv)
3561 {
3562   return (GNUNET_OK ==
3563           GNUNET_SERVICE_run (argc,
3564                               argv,
3565                               "fs",
3566                               GNUNET_SERVICE_OPTION_NONE,
3567                               &run, NULL)) ? 0 : 1;
3568 }
3569
3570 /* end of gnunet-service-fs.c */