plane hacking
[oweals/gnunet.git] / src / fs / gnunet-service-fs.c
index c5a47973c2d6b3efe51a4cdfeb01a0926c551688..9fce6478c2e93b0a0bb0bee5807a6ecb1956d8f9 100644 (file)
@@ -4,7 +4,7 @@
 
      GNUnet is free software; you can redistribute it and/or modify
      it under the terms of the GNU General Public License as published
-     by the Free Software Foundation; either version 2, or (at your
+     by the Free Software Foundation; either version 3, or (at your
      option) any later version.
 
      GNUnet is distributed in the hope that it will be useful, but
  * @brief gnunet anonymity protocol implementation
  * @author Christian Grothoff
  *
- * FIXME:
- * - TTL/priority calculations are absent!
  * TODO:
- * - have non-zero preference / priority for requests we initiate!
- * - implement hot-path routing decision procedure
- * - implement: bound_priority, test_load_too_high
- * - statistics
+ * - track per-peer request latency (using new load API)
+ * - consider more precise latency estimation (per-peer & request) -- again load API?
+ * - implement test_load_too_high, make decision priority-based, implement forwarding, etc.
+ * - introduce random latency in processing
+ * - tell other peers to stop migration if our PUTs fail (or if
+ *   we don't support migration per configuration?)
+ * - more statistics
  */
 #include "platform.h"
 #include <float.h>
 #include "gnunet_constants.h"
 #include "gnunet_core_service.h"
+#include "gnunet_dht_service.h"
 #include "gnunet_datastore_service.h"
+#include "gnunet_load_lib.h"
 #include "gnunet_peer_lib.h"
 #include "gnunet_protocols.h"
 #include "gnunet_signatures.h"
 #include "gnunet-service-fs_indexing.h"
 #include "fs.h"
 
-#define DEBUG_FS GNUNET_YES
+#define DEBUG_FS GNUNET_NO
 
 /**
  * Maximum number of outgoing messages we queue per peer.
- * FIXME: make configurable?
  */
 #define MAX_QUEUE_PER_PEER 16
 
+/**
+ * Size for the hash map for DHT requests from the FS
+ * service.  Should be about the number of concurrent
+ * DHT requests we plan to make.
+ */
+#define FS_DHT_HT_SIZE 1024
+
+/**
+ * How often do we flush trust values to disk?
+ */
+#define TRUST_FLUSH_FREQ GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MINUTES, 5)
+
 /**
  * Inverse of the probability that we will submit the same query
  * to the same peer again.  If the same peer already got the query
  */
 #define MAX_TRANSMIT_DELAY GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 45)
 
-
-
 /**
  * Maximum number of requests (from other peers) that we're
  * willing to have pending at any given point in time.
- * FIXME: set from configuration.
  */
-static uint64_t max_pending_requests = (32 * 1024);
+static unsigned long long max_pending_requests = (32 * 1024);
 
 
 /**
@@ -86,12 +97,6 @@ static uint64_t max_pending_requests = (32 * 1024);
  */
 struct PendingMessage;
 
-/**
- * Our connection to the datastore.
- */
-static struct GNUNET_DATASTORE_Handle *dsh;
-
-
 /**
  * Function called upon completion of a transmission.
  *
@@ -178,6 +183,12 @@ struct ConnectedPeer
    */ 
   struct GNUNET_TIME_Relative avg_delay;
 
+  /**
+   * Point in time until which this peer does not want us to migrate content
+   * to it.
+   */
+  struct GNUNET_TIME_Absolute migration_blocked;
+
   /**
    * Handle for an active request for transmission to this
    * peer, or NULL.
@@ -204,9 +215,19 @@ struct ConnectedPeer
 
   /**
    * Increase in traffic preference still to be submitted
-   * to the core service for this peer. FIXME: double or 'uint64_t'?
+   * to the core service for this peer.
    */
-  double inc_preference;
+  uint64_t inc_preference;
+
+  /**
+   * Trust rating for this peer
+   */
+  uint32_t trust;
+
+  /**
+   * Trust rating for this peer on disk.
+   */
+  uint32_t disk_trust;
 
   /**
    * The peer's identity.
@@ -480,7 +501,6 @@ struct PendingRequest
   struct GNUNET_DATASTORE_QueueEntry *qe;
 
   /**
-
    * Size of the 'bf' (in bytes).
    */
   size_t bf_size;
@@ -607,6 +627,21 @@ struct MigrationReadyBlock
 };
 
 
+/**
+ * Our connection to the datastore.
+ */
+static struct GNUNET_DATASTORE_Handle *dsh;
+
+/**
+ * Our block context.
+ */
+static struct GNUNET_BLOCK_Context *block_ctx;
+
+/**
+ * Our block configuration.
+ */
+static struct GNUNET_CONFIGURATION_Handle *block_cfg;
+
 /**
  * Our scheduler.
  */
@@ -673,6 +708,11 @@ static struct MigrationReadyBlock *mig_tail;
  */
 static struct GNUNET_DATASTORE_QueueEntry *mig_qe;
 
+/**
+ * Where do we store trust information?
+ */
+static char *trustDirectory;
+
 /**
  * ID of task that collects blocks for migration.
  */
@@ -684,6 +724,11 @@ static GNUNET_SCHEDULER_TaskIdentifier mig_task;
  */
 static struct GNUNET_TIME_Relative min_migration_delay;
 
+/**
+ * Handle for DHT operations.
+ */
+static struct GNUNET_DHT_Handle *dht_handle;
+
 /**
  * Size of the doubly-linked list of migration blocks.
  */
@@ -694,6 +739,58 @@ static unsigned int mig_size;
  */
 static int active_migration;
 
+/**
+ * Typical priorities we're seeing from other peers right now.  Since
+ * most priorities will be zero, this value is the weighted average of
+ * non-zero priorities seen "recently".  In order to ensure that new
+ * values do not dramatically change the ratio, values are first
+ * "capped" to a reasonable range (+N of the current value) and then
+ * averaged into the existing value by a ratio of 1:N.  Hence
+ * receiving the largest possible priority can still only raise our
+ * "current_priorities" by at most 1.
+ */
+static double current_priorities;
+
+/**
+ * Datastore load tracking.
+ */
+static struct GNUNET_LOAD_Value *datastore_load;
+
+
+/**
+ * We've just now completed a datastore request.  Update our
+ * datastore load calculations.
+ *
+ * @param start time when the datastore request was issued
+ */
+static void
+update_datastore_delays (struct GNUNET_TIME_Absolute start)
+{
+  struct GNUNET_TIME_Relative delay;
+
+  delay = GNUNET_TIME_absolute_get_duration (start);
+  GNUNET_LOAD_update (datastore_load,
+                     delay.value);
+}
+
+
+/**
+ * Get the filename under which we would store the GNUNET_HELLO_Message
+ * for the given host and protocol.
+ * @return filename of the form DIRECTORY/HOSTID
+ */
+static char *
+get_trust_filename (const struct GNUNET_PeerIdentity *id)
+{
+  struct GNUNET_CRYPTO_HashAsciiEncoded fil;
+  char *fn;
+
+  GNUNET_CRYPTO_hash_to_enc (&id->hashPubKey, &fil);
+  GNUNET_asprintf (&fn, "%s%s%s", trustDirectory, DIR_SEPARATOR_STR, &fil);
+  return fn;
+}
+
+
 
 /**
  * Transmit messages by copying it to the target buffer
@@ -715,7 +812,6 @@ transmit_to_peer (void *cls,
 
 /* ******************* clean up functions ************************ */
 
-
 /**
  * Delete the given migration block.
  *
@@ -760,7 +856,7 @@ is_closer (const GNUNET_HashCode *key,
  *            targets for (or NULL for none)
  * @param key ID of the peer 
  * @param value 'struct ConnectedPeer' of the peer
- * @return GNUNET_YES (always continue iteration)2
+ * @return GNUNET_YES (always continue iteration)
  */
 static int
 consider_migration (void *cls,
@@ -778,6 +874,8 @@ consider_migration (void *cls,
   unsigned int repl;
   
   /* consider 'cp' as a migration target for mb */
+  if (GNUNET_TIME_absolute_get_remaining (cp->migration_blocked).value > 0)
+    return GNUNET_YES; /* peer has requested no migration! */
   if (mb != NULL)
     {
       GNUNET_PEER_resolve (cp->pid,
@@ -842,11 +940,12 @@ consider_migration (void *cls,
     }
   if (msize == 0)
     return GNUNET_YES; /* no content available */
+#if DEBUG_FS
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-             "Trying to migrate `%s' (%u bytes) to `%s'\n",
-             GNUNET_h2s (&mb->query),
+             "Trying to migrate at least %u bytes to peer `%s'\n",
              msize,
-             GNUNET_i2s (&cppid));
+             GNUNET_h2s (key));
+#endif
   cp->cth 
     = GNUNET_CORE_notify_transmit_ready (core,
                                         0, GNUNET_TIME_UNIT_FOREVER_REL,
@@ -879,6 +978,8 @@ consider_migration_gathering ()
 {
   struct GNUNET_TIME_Relative delay;
 
+  if (dsh == NULL)
+    return;
   if (mig_qe != NULL)
     return;
   if (mig_task != GNUNET_SCHEDULER_NO_TASK)
@@ -930,7 +1031,7 @@ process_migration_content (void *cls,
        consider_migration_gathering ();
       return;
     }
-  if (type == GNUNET_BLOCK_TYPE_ONDEMAND)
+  if (type == GNUNET_BLOCK_TYPE_FS_ONDEMAND)
     {
       if (GNUNET_OK !=
          GNUNET_FS_handle_on_demand_block (key, size, data,
@@ -938,7 +1039,9 @@ process_migration_content (void *cls,
                                            expiration, uid, 
                                            &process_migration_content,
                                            NULL))
-       GNUNET_DATASTORE_get_next (dsh, GNUNET_YES);
+       {
+         GNUNET_DATASTORE_get_next (dsh, GNUNET_YES);
+       }
       return;
     }
 #if DEBUG_FS
@@ -977,10 +1080,13 @@ gather_migration_blocks (void *cls,
                         const struct GNUNET_SCHEDULER_TaskContext *tc)
 {
   mig_task = GNUNET_SCHEDULER_NO_TASK;
-  mig_qe = GNUNET_DATASTORE_get_random (dsh, 0, -1,
-                                       GNUNET_TIME_UNIT_FOREVER_REL,
-                                       &process_migration_content, NULL);
-  GNUNET_assert (mig_qe != NULL);
+  if (dsh != NULL)
+    {
+      mig_qe = GNUNET_DATASTORE_get_random (dsh, 0, -1,
+                                           GNUNET_TIME_UNIT_FOREVER_REL,
+                                           &process_migration_content, NULL);
+      GNUNET_assert (mig_qe != NULL);
+    }
 }
 
 
@@ -1141,9 +1247,18 @@ peer_connect_handler (void *cls,
 {
   struct ConnectedPeer *cp;
   struct MigrationReadyBlock *pos;
+  char *fn;
+  uint32_t trust;
   
   cp = GNUNET_malloc (sizeof (struct ConnectedPeer));
   cp->pid = GNUNET_PEER_intern (peer);
+
+  fn = get_trust_filename (peer);
+  if ((GNUNET_DISK_file_test (fn) == GNUNET_YES) &&
+      (sizeof (trust) == GNUNET_DISK_fn_read (fn, &trust, sizeof (trust))))
+    cp->disk_trust = cp->trust = ntohl (trust);
+  GNUNET_free (fn);
+
   GNUNET_break (GNUNET_OK ==
                GNUNET_CONTAINER_multihashmap_put (connected_peers,
                                                   &peer->hashPubKey,
@@ -1159,6 +1274,105 @@ peer_connect_handler (void *cls,
 }
 
 
+/**
+ * Increase the host credit by a value.
+ *
+ * @param host which peer to change the trust value on
+ * @param value is the int value by which the
+ *  host credit is to be increased or decreased
+ * @returns the actual change in trust (positive or negative)
+ */
+static int
+change_host_trust (struct ConnectedPeer *host, int value)
+{
+  unsigned int old_trust;
+
+  if (value == 0)
+    return 0;
+  GNUNET_assert (host != NULL);
+  old_trust = host->trust;
+  if (value > 0)
+    {
+      if (host->trust + value < host->trust)
+        {
+          value = UINT32_MAX - host->trust;
+          host->trust = UINT32_MAX;
+        }
+      else
+        host->trust += value;
+    }
+  else
+    {
+      if (host->trust < -value)
+        {
+          value = -host->trust;
+          host->trust = 0;
+        }
+      else
+        host->trust += value;
+    }
+  return value;
+}
+
+
+/**
+ * Write host-trust information to a file - flush the buffer entry!
+ */
+static int
+flush_trust (void *cls,
+            const GNUNET_HashCode *key,
+            void *value)
+{
+  struct ConnectedPeer *host = value;
+  char *fn;
+  uint32_t trust;
+  struct GNUNET_PeerIdentity pid;
+
+  if (host->trust == host->disk_trust)
+    return GNUNET_OK;                     /* unchanged */
+  GNUNET_PEER_resolve (host->pid,
+                      &pid);
+  fn = get_trust_filename (&pid);
+  if (host->trust == 0)
+    {
+      if ((0 != UNLINK (fn)) && (errno != ENOENT))
+        GNUNET_log_strerror_file (GNUNET_ERROR_TYPE_WARNING |
+                                  GNUNET_ERROR_TYPE_BULK, "unlink", fn);
+    }
+  else
+    {
+      trust = htonl (host->trust);
+      if (sizeof(uint32_t) == GNUNET_DISK_fn_write (fn, &trust, 
+                                                   sizeof(uint32_t),
+                                                   GNUNET_DISK_PERM_USER_READ | GNUNET_DISK_PERM_USER_WRITE
+                                                   | GNUNET_DISK_PERM_GROUP_READ | GNUNET_DISK_PERM_OTHER_READ))
+        host->disk_trust = host->trust;
+    }
+  GNUNET_free (fn);
+  return GNUNET_OK;
+}
+
+/**
+ * Call this method periodically to scan data/hosts for new hosts.
+ */
+static void
+cron_flush_trust (void *cls,
+                 const struct GNUNET_SCHEDULER_TaskContext *tc)
+{
+
+  if (NULL == connected_peers)
+    return;
+  GNUNET_CONTAINER_multihashmap_iterate (connected_peers,
+                                        &flush_trust,
+                                        NULL);
+  if (NULL == tc)
+    return;
+  if (0 != (tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN))
+    return;
+  GNUNET_SCHEDULER_add_delayed (tc->sched,
+                               TRUST_FLUSH_FREQ, &cron_flush_trust, NULL);
+}
+
 
 /**
  * Free (each) request made by the peer.
@@ -1246,7 +1460,6 @@ peer_disconnect_handler (void *cls,
                                             &consider_migration,
                                             pos);
     }
-
   GNUNET_PEER_change_rc (cp->pid, -1);
   GNUNET_PEER_decrement_rcs (cp->last_p2p_replies, P2P_SUCCESS_LIST_SIZE);
   if (NULL != cp->cth)
@@ -1389,6 +1602,7 @@ shutdown_task (void *cls,
   while (client_list != NULL)
     handle_client_disconnect (NULL,
                              client_list->client);
+  cron_flush_trust (NULL, NULL);
   GNUNET_CONTAINER_multihashmap_iterate (connected_peers,
                                         &clean_peer,
                                         NULL);
@@ -1411,14 +1625,27 @@ shutdown_task (void *cls,
       GNUNET_STATISTICS_destroy (stats, GNUNET_NO);
       stats = NULL;
     }
-  GNUNET_DATASTORE_disconnect (dsh,
-                              GNUNET_NO);
+  if (dsh != NULL)
+    {
+      GNUNET_DATASTORE_disconnect (dsh,
+                                  GNUNET_NO);
+      dsh = NULL;
+    }
   while (mig_head != NULL)
     delete_migration_block (mig_head);
   GNUNET_assert (0 == mig_size);
-  dsh = NULL;
+  GNUNET_DHT_disconnect (dht_handle);
+  dht_handle = NULL;
+  GNUNET_LOAD_value_free (datastore_load);
+  datastore_load = NULL;
+  GNUNET_BLOCK_context_destroy (block_ctx);
+  block_ctx = NULL;
+  GNUNET_CONFIGURATION_destroy (block_cfg);
+  block_cfg = NULL;
   sched = NULL;
   cfg = NULL;  
+  GNUNET_free_non_null (trustDirectory);
+  trustDirectory = NULL;
 }
 
 
@@ -1496,6 +1723,7 @@ transmit_to_peer (void *cls,
                  GNUNET_PEER_change_rc (mb->target_list[i], -1);
                  mb->target_list[i] = 0;
                  mb->used_targets++;
+                 memset (&migm, 0, sizeof (migm));
                  migm.header.size = htons (sizeof (migm) + mb->size);
                  migm.header.type = htons (GNUNET_MESSAGE_TYPE_FS_PUT);
                  migm.type = htonl (mb->type);
@@ -1609,34 +1837,17 @@ add_to_pending_messages_for_peer (struct ConnectedPeer *cp,
 }
 
 
-/**
- * Mingle hash with the mingle_number to produce different bits.
- */
-static void
-mingle_hash (const GNUNET_HashCode * in,
-            int32_t mingle_number, 
-            GNUNET_HashCode * hc)
-{
-  GNUNET_HashCode m;
-
-  GNUNET_CRYPTO_hash (&mingle_number, 
-                     sizeof (int32_t), 
-                     &m);
-  GNUNET_CRYPTO_hash_xor (&m, in, hc);
-}
-
-
 /**
  * Test if the load on this peer is too high
  * to even consider processing the query at
  * all.
  * 
- * @return GNUNET_YES if the load is too high, GNUNET_NO otherwise
+ * @return GNUNET_YES if the load is too high to do anything, GNUNET_NO to forward (load high, but not too high), GNUNET_SYSERR to indirect (load low)
  */
 static int
 test_load_too_high ()
 {
-  return GNUNET_NO; // FIXME
+  return GNUNET_SYSERR; // FIXME
 }
 
 
@@ -1788,7 +1999,9 @@ refresh_bloomfilter (struct PendingRequest *pr)
                                              BLOOMFILTER_K);
   for (i=0;i<pr->replies_seen_off;i++)
     {
-      mingle_hash (&pr->replies_seen[i], pr->mingle, &mhash);
+      GNUNET_BLOCK_mingle_hash (&pr->replies_seen[i],
+                               pr->mingle,
+                               &mhash);
       GNUNET_CONTAINER_bloomfilter_add (pr->bf, &mhash);
     }
 }
@@ -1855,11 +2068,6 @@ target_reservation_cb (void *cls,
       return;
     }
   no_route = GNUNET_NO;
-  /* FIXME: check against DBLOCK_SIZE and possibly return
-     amount to reserve; however, this also needs to work
-     with testcases which currently start out with a far
-     too low per-peer bw limit, so they would never send
-     anything.  Big issue. */
   if (amount == 0)
     {
       if (pr->cp == NULL)
@@ -1996,7 +2204,13 @@ target_peer_select_cb (void *cls,
 
   /* 1) check that this peer is not the initiator */
   if (cp == pr->cp)
-    return GNUNET_YES; /* skip */         
+    {
+#if DEBUG_FS
+      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                 "Skipping initiator in forwarding selection\n");
+#endif
+      return GNUNET_YES; /* skip */       
+    }
 
   /* 2) check if we have already (recently) forwarded to this peer */
   pc = 0;
@@ -2021,13 +2235,50 @@ target_peer_select_cb (void *cls,
                "Re-trying query that was previously transmitted %u times to this peer\n",
                (unsigned int) pc);
 #endif
-  // 3) calculate how much we'd like to forward to this peer
-  score = 42; // FIXME!
-  // FIXME: also need API to gather data on responsiveness
-  // of this peer (we have fields for that in 'cp', but
-  // they are never set!)
-  
+  /* 3) calculate how much we'd like to forward to this peer,
+     starting with a random value that is strong enough
+     to at least give any peer a chance sometimes 
+     (compared to the other factors that come later) */
+  /* 3a) count successful (recent) routes from cp for same source */
+  if (pr->cp != NULL)
+    {
+      score = GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
+                                       P2P_SUCCESS_LIST_SIZE);
+      for (i=0;i<P2P_SUCCESS_LIST_SIZE;i++)
+       if (cp->last_p2p_replies[i] == pr->cp->pid)
+         score += 1; /* likely successful based on hot path */
+    }
+  else
+    {
+      score = GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
+                                       CS2P_SUCCESS_LIST_SIZE);
+      for (i=0;i<CS2P_SUCCESS_LIST_SIZE;i++)
+       if (cp->last_client_replies[i] == pr->client_request_list->client_list->client)
+         score += 1; /* likely successful based on hot path */
+    }
+  /* 3b) include latency */
+  if (cp->avg_delay.value < 4 * TTL_DECREMENT)
+    score += 1; /* likely fast based on latency */
+  /* 3c) include priorities */
+  if (cp->avg_priority <= pr->remaining_priority / 2.0)
+    score += 1; /* likely successful based on priorities */
+  /* 3d) penalize for queue size */  
+  score -= (2.0 * cp->pending_requests / (double) MAX_QUEUE_PER_PEER); 
+  /* 3e) include peer proximity */
+  score -= (2.0 * (GNUNET_CRYPTO_hash_distance_u32 (key,
+                                                   &pr->query)) / (double) UINT32_MAX);
+  /* 4) super-bonus for being the known target */
+  if (pr->target_pid == cp->pid)
+    score += 100.0;
   /* store best-fit in closure */
+#if DEBUG_FS
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+             "Peer `%s' gets score %f for forwarding query, max is %f\n",
+             GNUNET_h2s (key),
+             score,
+             psc->target_score);
+#endif  
+  score++; /* avoid zero */
   if (score > psc->target_score)
     {
       psc->target_score = score;
@@ -2065,7 +2316,7 @@ bound_ttl (int32_t ttl_in, uint32_t prio)
 
 
 /**
- * We're processing a GET request from another peer and have decided
+ * We're processing a GET request and have decided
  * to forward it to other peers.  This function is called periodically
  * and should forward the request to other peers until we have all
  * possible replies.  If we have transmitted the *only* reply to
@@ -2097,13 +2348,28 @@ forward_request_task (void *cls,
     }
   if (GNUNET_YES == pr->local_only)
     return; /* configured to not do P2P search */
+  /* (0) try DHT */
+  if (0 == pr->anonymity_level)
+    {
+#if 0      
+      /* DHT API needs fixing... */
+      pr->dht_get = GNUNET_DHT_get_start (dht_handle,
+                                         GNUNET_TIME_UNIT_FOREVER_REL,
+                                         pr->type,
+                                         &pr->query,
+                                         &process_dht_reply,
+                                         pr,
+                                         FIXME,
+                                         FIXME);
+#endif                                   
+    }
   /* (1) select target */
   psc.pr = pr;
-  psc.target_score = DBL_MIN;
+  psc.target_score = -DBL_MAX;
   GNUNET_CONTAINER_multihashmap_iterate (connected_peers,
                                         &target_peer_select_cb,
                                         &psc);  
-  if (psc.target_score == DBL_MIN)
+  if (psc.target_score == -DBL_MAX)
     {
       delay = get_processing_delay ();
 #if DEBUG_FS 
@@ -2119,15 +2385,18 @@ forward_request_task (void *cls,
       return; /* nobody selected */
     }
   /* (3) update TTL/priority */
-  
   if (pr->client_request_list != NULL)
     {
       /* FIXME: use better algorithm!? */
       if (0 == GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
                                         4))
        pr->priority++;
-      /* FIXME: bound priority by "customary" priority used by other peers
-        at this time! */
+      /* bound priority we use by priorities we see from other peers
+        rounded up (must round up so that we can see non-zero
+        priorities, but round up as little as possible to make it
+        plausible that we forwarded another peers request) */
+      if (pr->priority > current_priorities + 1.0)
+       pr->priority = (uint32_t) current_priorities + 1.0;
       pr->ttl = bound_ttl (pr->ttl + TTL_DECREMENT * 2,
                           pr->priority);
 #if DEBUG_FS
@@ -2138,10 +2407,6 @@ forward_request_task (void *cls,
                  pr->ttl);
 #endif
     }
-  else
-    {
-      /* FIXME: should we do something here as well!? */
-    }
 
   /* (3) reserve reply bandwidth */
   cp = GNUNET_CONTAINER_multihashmap_get (connected_peers,
@@ -2150,12 +2415,12 @@ forward_request_task (void *cls,
   pr->irc = GNUNET_CORE_peer_change_preference (sched, cfg,
                                                &psc.target,
                                                GNUNET_CONSTANTS_SERVICE_TIMEOUT, 
-                                               GNUNET_BANDWIDTH_value_init ((uint32_t) -1 /* no limit */), 
+                                               GNUNET_BANDWIDTH_value_init (UINT32_MAX),
                                                DBLOCK_SIZE * 2, 
-                                               (uint64_t) cp->inc_preference,
+                                               cp->inc_preference,
                                                &target_reservation_cb,
                                                pr);
-  cp->inc_preference = 0.0;
+  cp->inc_preference = 0;
 }
 
 
@@ -2177,14 +2442,14 @@ transmit_reply_continuation (void *cls,
   
   switch (pr->type)
     {
-    case GNUNET_BLOCK_TYPE_DBLOCK:
-    case GNUNET_BLOCK_TYPE_IBLOCK:
+    case GNUNET_BLOCK_TYPE_FS_DBLOCK:
+    case GNUNET_BLOCK_TYPE_FS_IBLOCK:
       /* only one reply expected, done with the request! */
       destroy_pending_request (pr);
       break;
     case GNUNET_BLOCK_TYPE_ANY:
-    case GNUNET_BLOCK_TYPE_KBLOCK:
-    case GNUNET_BLOCK_TYPE_SBLOCK:
+    case GNUNET_BLOCK_TYPE_FS_KBLOCK:
+    case GNUNET_BLOCK_TYPE_FS_SBLOCK:
       break;
     default:
       GNUNET_break (0);
@@ -2276,12 +2541,6 @@ struct ProcessReplyClosure
    */
   size_t size;
 
-  /**
-   * Namespace that this reply belongs to
-   * (if it is of type SBLOCK).
-   */
-  GNUNET_HashCode namespace;
-
   /**
    * Type of the block.
    */
@@ -2292,6 +2551,11 @@ struct ProcessReplyClosure
    */
   uint32_t priority;
 
+  /**
+   * Evaluation result (returned).
+   */
+  enum GNUNET_BLOCK_EvaluationResult eval;
+
   /**
    * Did we finish processing the associated request?
    */ 
@@ -2320,8 +2584,6 @@ process_reply (void *cls,
   struct PutMessage *pm;
   struct ConnectedPeer *cp;
   struct GNUNET_TIME_Relative cur_delay;
-  GNUNET_HashCode chash;
-  GNUNET_HashCode mhash;
   size_t msize;
 
 #if DEBUG_FS
@@ -2348,7 +2610,8 @@ process_reply (void *cls,
       if (pr->cp != NULL)
        {
          GNUNET_PEER_change_rc (prq->sender->last_p2p_replies
-                                [prq->sender->last_p2p_replies_woff % P2P_SUCCESS_LIST_SIZE], -1);
+                                [prq->sender->last_p2p_replies_woff % P2P_SUCCESS_LIST_SIZE], 
+                                -1);
          GNUNET_PEER_change_rc (pr->cp->pid, 1);
          prq->sender->last_p2p_replies
            [(prq->sender->last_p2p_replies_woff++) % P2P_SUCCESS_LIST_SIZE]
@@ -2366,14 +2629,19 @@ process_reply (void *cls,
          GNUNET_SERVER_client_keep (pr->client_request_list->client_list->client);
        }
     }
-  GNUNET_CRYPTO_hash (prq->data,
-                     prq->size,
-                     &chash);
-  switch (prq->type)
-    {
-    case GNUNET_BLOCK_TYPE_DBLOCK:
-    case GNUNET_BLOCK_TYPE_IBLOCK:
-      /* only possible reply, stop requesting! */
+  prq->eval = GNUNET_BLOCK_evaluate (block_ctx,
+                                    prq->type,
+                                    key,
+                                    &pr->bf,
+                                    pr->mingle,
+                                    pr->namespace, (pr->namespace != NULL) ? sizeof (GNUNET_HashCode) : 0,
+                                    prq->data,
+                                    prq->size);
+  switch (prq->eval)
+    {
+    case GNUNET_BLOCK_EVALUATION_OK_MORE:
+      break;
+    case GNUNET_BLOCK_EVALUATION_OK_LAST:
       while (NULL != pr->pending_head)
        destroy_pending_message_list_entry (pr->pending_head);
       if (pr->qe != NULL)
@@ -2381,7 +2649,7 @@ process_reply (void *cls,
          if (pr->client_request_list != NULL)
            GNUNET_SERVER_receive_done (pr->client_request_list->client_list->client, 
                                        GNUNET_YES);
-         GNUNET_DATASTORE_cancel (pr->qe);
+         GNUNET_DATASTORE_cancel (pr->qe);
          pr->qe = NULL;
        }
       pr->do_remove = GNUNET_YES;
@@ -2396,66 +2664,57 @@ process_reply (void *cls,
                                                          key,
                                                          pr));
       break;
-    case GNUNET_BLOCK_TYPE_SBLOCK:
-      if (pr->namespace == NULL)
-       {
-         GNUNET_break (0);
-         return GNUNET_YES;
-       }
-      if (0 != memcmp (pr->namespace,
-                      &prq->namespace,
-                      sizeof (GNUNET_HashCode)))
-       {
-         GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
-                     _("Reply mismatched in terms of namespace.  Discarded.\n"));
-         return GNUNET_YES; /* wrong namespace */      
-       }
-      /* then: fall-through! */
-    case GNUNET_BLOCK_TYPE_KBLOCK:
-    case GNUNET_BLOCK_TYPE_NBLOCK:
-      if (pr->bf != NULL) 
-       {
-         mingle_hash (&chash, pr->mingle, &mhash);
-         if (GNUNET_YES == GNUNET_CONTAINER_bloomfilter_test (pr->bf,
-                                                              &mhash))
-           {
-             GNUNET_STATISTICS_update (stats,
-                                       gettext_noop ("# duplicate replies discarded (bloomfilter)"),
-                                       1,
-                                       GNUNET_NO);
-#if DEBUG_FS
-             GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-                         "Duplicate response `%s', discarding.\n",
-                         GNUNET_h2s (&mhash));
-#endif
-             return GNUNET_YES; /* duplicate */
-           }
+    case GNUNET_BLOCK_EVALUATION_OK_DUPLICATE:
+      GNUNET_STATISTICS_update (stats,
+                               gettext_noop ("# duplicate replies discarded (bloomfilter)"),
+                               1,
+                               GNUNET_NO);
 #if DEBUG_FS
-         GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-                     "New response `%s', adding to filter.\n",
-                     GNUNET_h2s (&mhash));
+      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                 "Duplicate response `%s', discarding.\n",
+                 GNUNET_h2s (&mhash));
 #endif
-       }
-      if (pr->client_request_list != NULL)
-       {
-         if (pr->replies_seen_size == pr->replies_seen_off)
-           GNUNET_array_grow (pr->replies_seen,
-                              pr->replies_seen_size,
-                              pr->replies_seen_size * 2 + 4);  
-           pr->replies_seen[pr->replies_seen_off++] = chash;         
-       }
-      if ( (pr->bf == NULL) ||
-          (pr->client_request_list != NULL) )
-       refresh_bloomfilter (pr);
-      GNUNET_CONTAINER_bloomfilter_add (pr->bf,
-                                       &mhash);
-      break;
-    default:
+      return GNUNET_YES; /* duplicate */
+    case GNUNET_BLOCK_EVALUATION_RESULT_INVALID:
+      return GNUNET_YES; /* wrong namespace */ 
+    case GNUNET_BLOCK_EVALUATION_REQUEST_VALID:
+      GNUNET_break (0);
+      return GNUNET_YES;
+    case GNUNET_BLOCK_EVALUATION_REQUEST_INVALID:
       GNUNET_break (0);
       return GNUNET_YES;
+    case GNUNET_BLOCK_EVALUATION_TYPE_NOT_SUPPORTED:
+      GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
+                 _("Unsupported block type %u\n"),
+                 prq->type);
+      return GNUNET_NO;
+    }
+  if (pr->client_request_list != NULL)
+    {
+      if (pr->replies_seen_size == pr->replies_seen_off)
+       GNUNET_array_grow (pr->replies_seen,
+                          pr->replies_seen_size,
+                          pr->replies_seen_size * 2 + 4);      
+      GNUNET_CRYPTO_hash (prq->data,
+                         prq->size,
+                         &pr->replies_seen[pr->replies_seen_off++]);         
+      refresh_bloomfilter (pr);
+    }
+  if (NULL == prq->sender)
+    {
+#if DEBUG_FS
+      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                 "Found result for query `%s' in local datastore\n",
+                 GNUNET_h2s (key));
+#endif
+      GNUNET_STATISTICS_update (stats,
+                               gettext_noop ("# results found locally"),
+                               1,
+                               GNUNET_NO);      
     }
   prq->priority += pr->remaining_priority;
   pr->remaining_priority = 0;
+  pr->results_found++;
   if (NULL != pr->client_request_list)
     {
       GNUNET_STATISTICS_update (stats,
@@ -2515,7 +2774,7 @@ process_reply (void *cls,
       reply->cont = &transmit_reply_continuation;
       reply->cont_cls = pr;
       reply->msize = msize;
-      reply->priority = (uint32_t) -1; /* send replies first! */
+      reply->priority = UINT32_MAX; /* send replies first! */
       pm = (struct PutMessage*) &reply[1];
       pm->header.type = htons (GNUNET_MESSAGE_TYPE_FS_PUT);
       pm->header.size = htons (msize);
@@ -2571,7 +2830,6 @@ handle_p2p_put (void *cls,
   struct GNUNET_TIME_Absolute expiration;
   GNUNET_HashCode query;
   struct ProcessReplyClosure prq;
-  const struct SBlock *sb;
 
   msize = ntohs (message->size);
   if (msize < sizeof (struct PutMessage))
@@ -2584,25 +2842,18 @@ handle_p2p_put (void *cls,
   type = ntohl (put->type);
   expiration = GNUNET_TIME_absolute_ntoh (put->expiration);
 
+  if (type == GNUNET_BLOCK_TYPE_FS_ONDEMAND)
+    return GNUNET_SYSERR;
   if (GNUNET_OK !=
-      GNUNET_BLOCK_check_block (type,
-                               &put[1],
-                               dsize,
-                               &query))
+      GNUNET_BLOCK_get_key (block_ctx,
+                           type,
+                           &put[1],
+                           dsize,
+                           &query))
     {
       GNUNET_break_op (0);
       return GNUNET_SYSERR;
     }
-  if (type == GNUNET_BLOCK_TYPE_ONDEMAND)
-    return GNUNET_SYSERR;
-  if (GNUNET_BLOCK_TYPE_SBLOCK == type)
-    { 
-      sb = (const struct SBlock*) &put[1];
-      GNUNET_CRYPTO_hash (&sb->subspace,
-                         sizeof (struct GNUNET_CRYPTO_RsaPublicKeyBinaryEncoded),
-                         &prq.namespace);
-    }
-
 #if DEBUG_FS
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Received result for query `%s' from peer `%4s'\n",
@@ -2618,6 +2869,8 @@ handle_p2p_put (void *cls,
   if (other != NULL)
     prq.sender = GNUNET_CONTAINER_multihashmap_get (connected_peers,
                                                    &other->hashPubKey);
+  else
+    prq.sender = NULL;
   prq.size = dsize;
   prq.type = type;
   prq.expiration = expiration;
@@ -2627,6 +2880,11 @@ handle_p2p_put (void *cls,
                                              &query,
                                              &process_reply,
                                              &prq);
+  if (prq.sender != NULL)
+    {
+      prq.sender->inc_preference += CONTENT_BANDWIDTH_VALUE + 1000 * prq.priority;
+      prq.sender->trust += prq.priority;
+    }
   if (GNUNET_YES == active_migration)
     {
 #if DEBUG_FS
@@ -2648,6 +2906,30 @@ handle_p2p_put (void *cls,
 }
 
 
+/**
+ * Handle P2P "MIGRATION_STOP" message.
+ *
+ * @param cls closure, always NULL
+ * @param other the other peer involved (sender or receiver, NULL
+ *        for loopback messages where we are both sender and receiver)
+ * @param message the actual message
+ * @param latency reported latency of the connection with 'other'
+ * @param distance reported distance (DV) to 'other' 
+ * @return GNUNET_OK to keep the connection open,
+ *         GNUNET_SYSERR to close it (signal serious error)
+ */
+static int
+handle_p2p_migration_stop (void *cls,
+                          const struct GNUNET_PeerIdentity *other,
+                          const struct GNUNET_MessageHeader *message,
+                          struct GNUNET_TIME_Relative latency,
+                          uint32_t distance)
+{
+  // FIXME!
+}
+
+
+
 /* **************************** P2P GET Handling ************************ */
 
 
@@ -2733,10 +3015,8 @@ process_local_reply (void *cls,
   struct PendingRequest *pr = cls;
   struct ProcessReplyClosure prq;
   struct CheckDuplicateRequestClosure cdrc;
-  const struct SBlock *sb;
-  GNUNET_HashCode dhash;
-  GNUNET_HashCode mhash;
   GNUNET_HashCode query;
+  unsigned int old_rf;
   
   if (NULL == key)
     {
@@ -2783,7 +3063,7 @@ process_local_reply (void *cls,
              GNUNET_h2s (key),
              type);
 #endif
-  if (type == GNUNET_BLOCK_TYPE_ONDEMAND)
+  if (type == GNUNET_BLOCK_TYPE_FS_ONDEMAND)
     {
 #if DEBUG_FS
       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
@@ -2799,56 +3079,22 @@ process_local_reply (void *cls,
                                            &process_local_reply,
                                            pr))
       if (pr->qe != NULL)
-       GNUNET_DATASTORE_get_next (dsh, GNUNET_YES);
-      return;
-    }
-  /* check for duplicates */
-  GNUNET_CRYPTO_hash (data, size, &dhash);
-  mingle_hash (&dhash, 
-              pr->mingle,
-              &mhash);
-  if ( (pr->bf != NULL) &&
-       (GNUNET_YES ==
-       GNUNET_CONTAINER_bloomfilter_test (pr->bf,
-                                          &mhash)) )
-    {      
-#if DEBUG_FS
-      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-                 "Result from datastore filtered by bloomfilter (duplicate).\n");
-#endif
-      GNUNET_STATISTICS_update (stats,
-                               gettext_noop ("# results filtered by query bloomfilter"),
-                               1,
-                               GNUNET_NO);
-      if (pr->qe != NULL)
-       GNUNET_DATASTORE_get_next (dsh, GNUNET_YES);
+       {
+         GNUNET_DATASTORE_get_next (dsh, GNUNET_YES);
+       }
       return;
     }
-#if DEBUG_FS
-  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-             "Found result for query `%s' in local datastore\n",
-             GNUNET_h2s (key));
-#endif
-  GNUNET_STATISTICS_update (stats,
-                           gettext_noop ("# results found locally"),
-                           1,
-                           GNUNET_NO);
-  pr->results_found++;
+  old_rf = pr->results_found;
   memset (&prq, 0, sizeof (prq));
   prq.data = data;
   prq.expiration = expiration;
   prq.size = size;  
-  if (GNUNET_BLOCK_TYPE_SBLOCK == type)
-    { 
-      sb = (const struct SBlock*) data;
-      GNUNET_CRYPTO_hash (&sb->subspace,
-                         sizeof (struct GNUNET_CRYPTO_RsaPublicKeyBinaryEncoded),
-                         &prq.namespace);
-    }
-  if (GNUNET_OK != GNUNET_BLOCK_check_block (type,
-                                            data,
-                                            size,
-                                            &query))
+  if (GNUNET_OK != 
+      GNUNET_BLOCK_get_key (block_ctx,
+                           type,
+                           data,
+                           size,
+                           &query))
     {
       GNUNET_break (0);
       GNUNET_DATASTORE_remove (dsh,
@@ -2864,12 +3110,14 @@ process_local_reply (void *cls,
   prq.priority = priority;  
   prq.finished = GNUNET_NO;
   process_reply (&prq, key, pr);
+  if ( (old_rf == 0) &&
+       (pr->results_found == 1) )
+    update_datastore_delays (pr->start_time);
   if (prq.finished == GNUNET_YES)
     return;
   if (pr->qe == NULL)
     return; /* done here */
-  if ( (type == GNUNET_BLOCK_TYPE_DBLOCK) ||
-       (type == GNUNET_BLOCK_TYPE_IBLOCK) ) 
+  if (prq.eval == GNUNET_BLOCK_EVALUATION_OK_LAST)
     {
       GNUNET_DATASTORE_get_next (dsh, GNUNET_NO);
       return;
@@ -2905,7 +3153,26 @@ static uint32_t
 bound_priority (uint32_t prio_in,
                struct ConnectedPeer *cp)
 {
-  return 0; // FIXME!
+#define N ((double)128.0)
+  uint32_t ret;
+  double rret;
+  int ld;
+
+  ld = test_load_too_high ();
+  if (ld == GNUNET_SYSERR)
+    return 0; /* excess resources */
+  ret = change_host_trust (cp, prio_in);
+  if (ret > 0)
+    {
+      if (ret > current_priorities + N)
+       rret = current_priorities + N;
+      else
+       rret = ret;
+      current_priorities 
+       = (current_priorities * (N-1) + rret)/N;
+    }
+#undef N
+  return ret;
 }
 
 
@@ -2971,8 +3238,8 @@ handle_p2p_get (void *cls,
   size_t bfsize;
   uint32_t ttl_decrement;
   enum GNUNET_BLOCK_Type type;
-  double preference;
   int have_ns;
+  int ld;
 
   msize = ntohs(message->size);
   if (msize < sizeof (struct GetMessage))
@@ -2982,18 +3249,6 @@ handle_p2p_get (void *cls,
     }
   gm = (const struct GetMessage*) message;
   type = ntohl (gm->type);
-  switch (type)
-    {
-    case GNUNET_BLOCK_TYPE_ANY:
-    case GNUNET_BLOCK_TYPE_DBLOCK:
-    case GNUNET_BLOCK_TYPE_IBLOCK:
-    case GNUNET_BLOCK_TYPE_KBLOCK:
-    case GNUNET_BLOCK_TYPE_SBLOCK:
-      break;
-    default:
-      GNUNET_break_op (0);
-      return GNUNET_SYSERR;
-    }
   bm = ntohl (gm->hash_bitmap);
   bits = 0;
   while (bm > 0)
@@ -3010,12 +3265,6 @@ handle_p2p_get (void *cls,
   opt = (const GNUNET_HashCode*) &gm[1];
   bfsize = msize - sizeof (struct GetMessage) + bits * sizeof (GNUNET_HashCode);
   bm = ntohl (gm->hash_bitmap);
-  if ( (0 != (bm & GET_MESSAGE_BIT_SKS_NAMESPACE)) &&
-       (type != GNUNET_BLOCK_TYPE_SBLOCK) )
-    {
-      GNUNET_break_op (0);
-      return GNUNET_SYSERR;      
-    }
   bits = 0;
   cps = GNUNET_CONTAINER_multihashmap_get (connected_peers,
                                           &other->hashPubKey);
@@ -3056,7 +3305,11 @@ handle_p2p_get (void *cls,
   /* note that we can really only check load here since otherwise
      peers could find out that we are overloaded by not being
      disconnected after sending us a malformed query... */
-  if (GNUNET_YES == test_load_too_high ())
+
+  /* FIXME: query priority should play
+     a major role here! */
+  ld = test_load_too_high ();
+  if (GNUNET_YES == ld)
     {
 #if DEBUG_FS
       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
@@ -3069,6 +3322,8 @@ handle_p2p_get (void *cls,
                                GNUNET_NO);
       return GNUNET_OK;
     }
+  /* FIXME: if ld == GNUNET_NO, forward
+     instead of indirecting! */
 
 #if DEBUG_FS 
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
@@ -3082,14 +3337,14 @@ handle_p2p_get (void *cls,
   pr = GNUNET_malloc (sizeof (struct PendingRequest) + 
                      (have_ns ? sizeof(GNUNET_HashCode) : 0));
   if (have_ns)
-    pr->namespace = (GNUNET_HashCode*) &pr[1];
+    {
+      pr->namespace = (GNUNET_HashCode*) &pr[1];
+      memcpy (&pr[1], &opt[bits++], sizeof (GNUNET_HashCode));
+    }
   pr->type = type;
   pr->mingle = ntohl (gm->filter_mutator);
-  if (0 != (bm & GET_MESSAGE_BIT_SKS_NAMESPACE))    
-    memcpy (&pr[1], &opt[bits++], sizeof (GNUNET_HashCode));
   if (0 != (bm & GET_MESSAGE_BIT_TRANSMIT_TO))
     pr->target_pid = GNUNET_PEER_intern ((const struct GNUNET_PeerIdentity*) &opt[bits++]);
-
   pr->anonymity_level = 1;
   pr->priority = bound_priority (ntohl (gm->priority), cps);
   pr->ttl = bound_ttl (ntohl (gm->ttl), pr->priority);
@@ -3192,13 +3447,9 @@ handle_p2p_get (void *cls,
                            GNUNET_NO);
 
   /* calculate change in traffic preference */
-  preference = (double) pr->priority;
-  if (preference < QUERY_BANDWIDTH_VALUE)
-    preference = QUERY_BANDWIDTH_VALUE;
-  cps->inc_preference += preference;
-
+  cps->inc_preference += pr->priority * 1000 + QUERY_BANDWIDTH_VALUE;
   /* process locally */
-  if (type == GNUNET_BLOCK_TYPE_DBLOCK)
+  if (type == GNUNET_BLOCK_TYPE_FS_DBLOCK)
     type = GNUNET_BLOCK_TYPE_ANY; /* to get on-demand as well */
   timeout = GNUNET_TIME_relative_multiply (BASIC_DATASTORE_REQUEST_DELAY,
                                           (pr->priority + 1)); 
@@ -3214,8 +3465,8 @@ handle_p2p_get (void *cls,
   /* Are multiple results possible?  If so, start processing remotely now! */
   switch (pr->type)
     {
-    case GNUNET_BLOCK_TYPE_DBLOCK:
-    case GNUNET_BLOCK_TYPE_IBLOCK:
+    case GNUNET_BLOCK_TYPE_FS_DBLOCK:
+    case GNUNET_BLOCK_TYPE_FS_IBLOCK:
       /* only one result, wait for datastore */
       break;
     default:
@@ -3229,6 +3480,7 @@ handle_p2p_get (void *cls,
   if (GNUNET_CONTAINER_heap_get_size (requests_by_expiration_heap) > max_pending_requests)
     {
       pr = GNUNET_CONTAINER_heap_peek (requests_by_expiration_heap);
+      GNUNET_assert (pr != NULL);
       destroy_pending_request (pr);
     }
   return GNUNET_OK;
@@ -3281,22 +3533,6 @@ handle_start_search (void *cls,
              GNUNET_h2s (&sm->query),
              (unsigned int) type);
 #endif
-  switch (type)
-    {
-    case GNUNET_BLOCK_TYPE_ANY:
-    case GNUNET_BLOCK_TYPE_DBLOCK:
-    case GNUNET_BLOCK_TYPE_IBLOCK:
-    case GNUNET_BLOCK_TYPE_KBLOCK:
-    case GNUNET_BLOCK_TYPE_SBLOCK:
-    case GNUNET_BLOCK_TYPE_NBLOCK:
-      break;
-    default:
-      GNUNET_break (0);
-      GNUNET_SERVER_receive_done (client,
-                                 GNUNET_SYSERR);
-      return;
-    }  
-
   cl = client_list;
   while ( (cl != NULL) &&
          (cl->client != client) )
@@ -3310,8 +3546,8 @@ handle_start_search (void *cls,
       client_list = cl;
     }
   /* detect duplicate KBLOCK requests */
-  if ( (type == GNUNET_BLOCK_TYPE_KBLOCK) ||
-       (type == GNUNET_BLOCK_TYPE_NBLOCK) ||
+  if ( (type == GNUNET_BLOCK_TYPE_FS_KBLOCK) ||
+       (type == GNUNET_BLOCK_TYPE_FS_NBLOCK) ||
        (type == GNUNET_BLOCK_TYPE_ANY) )
     {
       crl = cl->rl_head;
@@ -3353,7 +3589,7 @@ handle_start_search (void *cls,
                            1,
                            GNUNET_NO);
   pr = GNUNET_malloc (sizeof (struct PendingRequest) + 
-                     ((type == GNUNET_BLOCK_TYPE_SBLOCK) ? sizeof(GNUNET_HashCode) : 0));
+                     ((type == GNUNET_BLOCK_TYPE_FS_SBLOCK) ? sizeof(GNUNET_HashCode) : 0));
   crl = GNUNET_malloc (sizeof (struct ClientRequestList));
   memset (crl, 0, sizeof (struct ClientRequestList));
   crl->client_list = cl;
@@ -3371,6 +3607,7 @@ handle_start_search (void *cls,
          sc * sizeof (GNUNET_HashCode));
   pr->replies_seen_off = sc;
   pr->anonymity_level = ntohl (sm->anonymity_level); 
+  pr->start_time = GNUNET_TIME_absolute_get ();
   refresh_bloomfilter (pr);
   pr->query = sm->query;
   if (0 == (1 & ntohl (sm->options)))
@@ -3379,14 +3616,14 @@ handle_start_search (void *cls,
     pr->local_only = GNUNET_YES;
   switch (type)
     {
-    case GNUNET_BLOCK_TYPE_DBLOCK:
-    case GNUNET_BLOCK_TYPE_IBLOCK:
+    case GNUNET_BLOCK_TYPE_FS_DBLOCK:
+    case GNUNET_BLOCK_TYPE_FS_IBLOCK:
       if (0 != memcmp (&sm->target,
                       &all_zeros,
                       sizeof (GNUNET_HashCode)))
        pr->target_pid = GNUNET_PEER_intern ((const struct GNUNET_PeerIdentity*) &sm->target);
       break;
-    case GNUNET_BLOCK_TYPE_SBLOCK:
+    case GNUNET_BLOCK_TYPE_FS_SBLOCK:
       pr->namespace = (GNUNET_HashCode*) &pr[1];
       memcpy (&pr[1], &sm->target, sizeof (GNUNET_HashCode));
       break;
@@ -3398,7 +3635,7 @@ handle_start_search (void *cls,
                                                   &sm->query,
                                                   pr,
                                                   GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE));
-  if (type == GNUNET_BLOCK_TYPE_DBLOCK)
+  if (type == GNUNET_BLOCK_TYPE_FS_DBLOCK)
     type = GNUNET_BLOCK_TYPE_ANY; /* get on-demand blocks too! */
   pr->qe = GNUNET_DATASTORE_get (dsh,
                                 &sm->query,
@@ -3412,38 +3649,6 @@ handle_start_search (void *cls,
 
 /* **************************** Startup ************************ */
 
-
-/**
- * List of handlers for P2P messages
- * that we care about.
- */
-static struct GNUNET_CORE_MessageHandler p2p_handlers[] =
-  {
-    { &handle_p2p_get, 
-      GNUNET_MESSAGE_TYPE_FS_GET, 0 },
-    { &handle_p2p_put, 
-      GNUNET_MESSAGE_TYPE_FS_PUT, 0 },
-    { NULL, 0, 0 }
-  };
-
-
-/**
- * List of handlers for the messages understood by this
- * service.
- */
-static struct GNUNET_SERVER_MessageHandler handlers[] = {
-  {&GNUNET_FS_handle_index_start, NULL, 
-   GNUNET_MESSAGE_TYPE_FS_INDEX_START, 0},
-  {&GNUNET_FS_handle_index_list_get, NULL, 
-   GNUNET_MESSAGE_TYPE_FS_INDEX_LIST_GET, sizeof(struct GNUNET_MessageHeader) },
-  {&GNUNET_FS_handle_unindex, NULL, GNUNET_MESSAGE_TYPE_FS_UNINDEX, 
-   sizeof (struct UnindexMessage) },
-  {&handle_start_search, NULL, GNUNET_MESSAGE_TYPE_FS_START_SEARCH, 
-   0 },
-  {NULL, NULL, 0, 0}
-};
-
-
 /**
  * Process fs requests.
  *
@@ -3456,13 +3661,56 @@ main_init (struct GNUNET_SCHEDULER_Handle *s,
           struct GNUNET_SERVER_Handle *server,
           const struct GNUNET_CONFIGURATION_Handle *c)
 {
+  static const struct GNUNET_CORE_MessageHandler p2p_handlers[] =
+    {
+      { &handle_p2p_get, 
+       GNUNET_MESSAGE_TYPE_FS_GET, 0 },
+      { &handle_p2p_put, 
+       GNUNET_MESSAGE_TYPE_FS_PUT, 0 },
+      { &handle_p2p_migration_stop, 
+       GNUNET_MESSAGE_TYPE_FS_MIGRATION_STOP,
+       sizeof (struct MigrationStopMessage) },
+      { NULL, 0, 0 }
+    };
+  static const struct GNUNET_SERVER_MessageHandler handlers[] = {
+    {&GNUNET_FS_handle_index_start, NULL, 
+     GNUNET_MESSAGE_TYPE_FS_INDEX_START, 0},
+    {&GNUNET_FS_handle_index_list_get, NULL, 
+     GNUNET_MESSAGE_TYPE_FS_INDEX_LIST_GET, sizeof(struct GNUNET_MessageHeader) },
+    {&GNUNET_FS_handle_unindex, NULL, GNUNET_MESSAGE_TYPE_FS_UNINDEX, 
+     sizeof (struct UnindexMessage) },
+    {&handle_start_search, NULL, GNUNET_MESSAGE_TYPE_FS_START_SEARCH, 
+     0 },
+    {NULL, NULL, 0, 0}
+  };
+  unsigned long long enc = 128;
+
   sched = s;
   cfg = c;
   stats = GNUNET_STATISTICS_create (sched, "fs", cfg);
-  min_migration_delay = GNUNET_TIME_UNIT_SECONDS; // FIXME: get from config
-  connected_peers = GNUNET_CONTAINER_multihashmap_create (128); // FIXME: get size from config
-  query_request_map = GNUNET_CONTAINER_multihashmap_create (128); // FIXME: get size from config
-  peer_request_map = GNUNET_CONTAINER_multihashmap_create (128); // FIXME: get size from config
+  min_migration_delay = GNUNET_TIME_UNIT_SECONDS;
+  if ( (GNUNET_OK !=
+       GNUNET_CONFIGURATION_get_value_number (cfg,
+                                              "fs",
+                                              "MAX_PENDING_REQUESTS",
+                                              &max_pending_requests)) ||
+       (GNUNET_OK !=
+       GNUNET_CONFIGURATION_get_value_number (cfg,
+                                              "fs",
+                                              "EXPECTED_NEIGHBOUR_COUNT",
+                                              &enc)) ||
+       (GNUNET_OK != 
+       GNUNET_CONFIGURATION_get_value_time (cfg,
+                                            "fs",
+                                            "MIN_MIGRATION_DELAY",
+                                            &min_migration_delay)) )
+    {
+      GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+                 _("Configuration fails to specify certain parameters, assuming default values."));
+    }
+  connected_peers = GNUNET_CONTAINER_multihashmap_create (enc); 
+  query_request_map = GNUNET_CONTAINER_multihashmap_create (max_pending_requests);
+  peer_request_map = GNUNET_CONTAINER_multihashmap_create (enc);
   requests_by_expiration_heap = GNUNET_CONTAINER_heap_create (GNUNET_CONTAINER_HEAP_ORDER_MIN); 
   core = GNUNET_CORE_connect (sched,
                              cfg,
@@ -3471,6 +3719,7 @@ main_init (struct GNUNET_SCHEDULER_Handle *s,
                              NULL,
                              &peer_connect_handler,
                              &peer_disconnect_handler,
+                             NULL,
                              NULL, GNUNET_NO,
                              NULL, GNUNET_NO,
                              p2p_handlers);
@@ -3504,6 +3753,17 @@ main_init (struct GNUNET_SCHEDULER_Handle *s,
   GNUNET_SERVER_disconnect_notify (server, 
                                   &handle_client_disconnect,
                                   NULL);
+  GNUNET_assert (GNUNET_OK ==
+                 GNUNET_CONFIGURATION_get_value_filename (cfg,
+                                                          "fs",
+                                                          "TRUST",
+                                                          &trustDirectory));
+  GNUNET_DISK_directory_create (trustDirectory);
+  GNUNET_SCHEDULER_add_with_priority (sched,
+                                     GNUNET_SCHEDULER_PRIORITY_HIGH,
+                                     &cron_flush_trust, NULL);
+
+
   GNUNET_SERVER_add_handlers (server, handlers);
   GNUNET_SCHEDULER_add_delayed (sched,
                                GNUNET_TIME_UNIT_FOREVER_REL,
@@ -3537,12 +3797,31 @@ run (void *cls,
       GNUNET_SCHEDULER_shutdown (sched);
       return;
     }
+  datastore_load = GNUNET_LOAD_value_init ();
+  block_cfg = GNUNET_CONFIGURATION_create ();
+  GNUNET_CONFIGURATION_set_value_string (block_cfg,
+                                        "block",
+                                        "PLUGINS",
+                                        "fs");
+  block_ctx = GNUNET_BLOCK_context_create (block_cfg);
+  GNUNET_assert (NULL != block_ctx);
+  dht_handle = GNUNET_DHT_connect (sched,
+                                  cfg,
+                                  FS_DHT_HT_SIZE);
   if ( (GNUNET_OK != GNUNET_FS_indexing_init (sched, cfg, dsh)) ||
        (GNUNET_OK != main_init (sched, server, cfg)) )
     {    
       GNUNET_SCHEDULER_shutdown (sched);
       GNUNET_DATASTORE_disconnect (dsh, GNUNET_NO);
       dsh = NULL;
+      GNUNET_DHT_disconnect (dht_handle);
+      dht_handle = NULL;
+      GNUNET_BLOCK_context_destroy (block_ctx);
+      block_ctx = NULL;
+      GNUNET_CONFIGURATION_destroy (block_cfg);
+      block_cfg = NULL;
+      GNUNET_LOAD_value_free (datastore_load);
+      datastore_load = NULL;
       return;   
     }
 }