more stats
[oweals/gnunet.git] / src / datastore / gnunet-service-datastore.c
index 89c5d6af8faef9d0fa6bee4b660339e031592211..c3370e0a582179a53787ab580c36c599adb9efcb 100644 (file)
  * @file datastore/gnunet-service-datastore.c
  * @brief Management for the datastore for files stored on a GNUnet node
  * @author Christian Grothoff
- *
- * TODO:
- * quota management code:
- * - track storage use
- * - refuse above-quota
- * - content expiration job
- * - near-quota low-priority content discard job
  */
 
 #include "platform.h"
 #include "gnunet_util_lib.h"
+#include "gnunet_arm_service.h"
 #include "gnunet_protocols.h"
+#include "gnunet_statistics_service.h"
 #include "plugin_datastore.h"
 #include "datastore.h"
 
  */
 #define MAX_PENDING 1024
 
+/**
+ * How long are we at most keeping "expired" content
+ * past the expiration date in the database?
+ */
+#define MAX_EXPIRE_DELAY GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MINUTES, 15)
+
+
 
 /**
  * Our datastore plugin.
@@ -133,11 +135,40 @@ static int reservation_gen;
  */
 static unsigned long long quota;
 
+/**
+ * How much space are we using for the cache?  (space available for
+ * insertions that will be instantly reclaimed by discarding less
+ * important content --- or possibly whatever we just inserted into
+ * the "cache").
+ */
+static unsigned long long cache_size;
+
 /**
  * How much space have we currently reserved?
  */
 static unsigned long long reserved;
 
+/**
+ * Identity of the task that is used to delete
+ * expired content.
+ */
+static GNUNET_SCHEDULER_TaskIdentifier expired_kill_task;
+
+/**
+ * Our configuration.
+ */
+const struct GNUNET_CONFIGURATION_Handle *cfg;
+
+/**
+ * Our scheduler.
+ */
+struct GNUNET_SCHEDULER_Handle *sched; 
+
+/**
+ * Handle for reporting statistics.
+ */
+static struct GNUNET_STATISTICS_Handle *stats;
+
 
 /**
  * Function called once the transmit operation has
@@ -149,12 +180,32 @@ static unsigned long long reserved;
 typedef void (*TransmitContinuation)(void *cls,
                                     int status);
 
+
+/**
+ * Context for transmitting replies to clients.
+ */
 struct TransmitCallbackContext 
 {
+  
+  /**
+   * We keep these in a doubly-linked list (for cleanup).
+   */
+  struct TransmitCallbackContext *next;
+  
+  /**
+   * We keep these in a doubly-linked list (for cleanup).
+   */
+  struct TransmitCallbackContext *prev;
+  
   /**
    * The message that we're asked to transmit.
    */
   struct GNUNET_MessageHeader *msg;
+  
+  /**
+   * Handle for the transmission request.
+   */
+  struct GNUNET_CONNECTION_TransmitHandle *th;
 
   /**
    * Client that we are transmitting to.
@@ -179,6 +230,217 @@ struct TransmitCallbackContext
   int end;
 };
 
+  
+/**
+ * Head of the doubly-linked list (for cleanup).
+ */
+static struct TransmitCallbackContext *tcc_head;
+
+/**
+ * Tail of the doubly-linked list (for cleanup).
+ */
+static struct TransmitCallbackContext *tcc_tail;
+
+/**
+ * Have we already clean ed up the TCCs and are hence no longer
+ * willing (or able) to transmit anything to anyone?
+ */
+static int cleaning_done;
+
+/**
+ * Task that is used to remove expired entries from
+ * the datastore.  This task will schedule itself
+ * again automatically to always delete all expired
+ * content quickly.
+ *
+ * @param cls not used
+ * @param tc task context
+ */ 
+static void
+delete_expired (void *cls,
+               const struct GNUNET_SCHEDULER_TaskContext *tc);
+
+
+/**
+ * Iterate over the expired items stored in the datastore.
+ * Delete all expired items; once we have processed all
+ * expired items, re-schedule the "delete_expired" task.
+ *
+ * @param cls not used
+ * @param next_cls closure to pass to the "next" function.
+ * @param key key for the content
+ * @param size number of bytes in data
+ * @param data content stored
+ * @param type type of the content
+ * @param priority priority of the content
+ * @param anonymity anonymity-level for the content
+ * @param expiration expiration time for the content
+ * @param uid unique identifier for the datum;
+ *        maybe 0 if no unique identifier is available
+ *
+ * @return GNUNET_SYSERR to abort the iteration, GNUNET_OK to continue
+ *         (continue on call to "next", of course),
+ *         GNUNET_NO to delete the item and continue (if supported)
+ */
+static int 
+expired_processor (void *cls,
+                  void *next_cls,
+                  const GNUNET_HashCode * key,
+                  uint32_t size,
+                  const void *data,
+                  uint32_t type,
+                  uint32_t priority,
+                  uint32_t anonymity,
+                  struct GNUNET_TIME_Absolute
+                  expiration, 
+                  uint64_t uid)
+{
+  struct GNUNET_TIME_Absolute now;
+
+  expired_kill_task = GNUNET_SCHEDULER_NO_TASK;
+  if (key == NULL) 
+    {
+      expired_kill_task 
+       = GNUNET_SCHEDULER_add_delayed (sched,
+                                       MAX_EXPIRE_DELAY,
+                                       &delete_expired,
+                                       NULL);
+      return GNUNET_SYSERR;
+    }
+  now = GNUNET_TIME_absolute_get ();
+  if (expiration.value > now.value)
+    {
+      /* finished processing */
+      plugin->api->next_request (next_cls, GNUNET_YES);
+      return GNUNET_SYSERR;
+    }
+  plugin->api->next_request (next_cls, GNUNET_NO);
+#if DEBUG_DATASTORE
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+             "Deleting content that expired %llu ms ago\n",
+             (unsigned long long) (now.value - expiration.value));
+#endif
+  GNUNET_STATISTICS_update (stats,
+                           gettext_noop ("# bytes expired"),
+                           size,
+                           GNUNET_NO);
+  GNUNET_CONTAINER_bloomfilter_remove (filter,
+                                      key);
+  return GNUNET_NO; /* delete */
+}
+
+
+/**
+ * Task that is used to remove expired entries from
+ * the datastore.  This task will schedule itself
+ * again automatically to always delete all expired
+ * content quickly.
+ *
+ * @param cls not used
+ * @param tc task context
+ */ 
+static void
+delete_expired (void *cls,
+               const struct GNUNET_SCHEDULER_TaskContext *tc)
+{
+  plugin->api->iter_ascending_expiration (plugin->api->cls, 
+                                         0,
+                                         &expired_processor,
+                                         NULL);
+}
+
+
+/**
+ * An iterator over a set of items stored in the datastore.
+ *
+ * @param cls closure
+ * @param next_cls closure to pass to the "next" function.
+ * @param key key for the content
+ * @param size number of bytes in data
+ * @param data content stored
+ * @param type type of the content
+ * @param priority priority of the content
+ * @param anonymity anonymity-level for the content
+ * @param expiration expiration time for the content
+ * @param uid unique identifier for the datum;
+ *        maybe 0 if no unique identifier is available
+ *
+ * @return GNUNET_SYSERR to abort the iteration, GNUNET_OK to continue
+ *         (continue on call to "next", of course),
+ *         GNUNET_NO to delete the item and continue (if supported)
+ */
+static int 
+manage (void *cls,
+       void *next_cls,
+       const GNUNET_HashCode * key,
+       uint32_t size,
+       const void *data,
+       uint32_t type,
+       uint32_t priority,
+       uint32_t anonymity,
+       struct GNUNET_TIME_Absolute
+       expiration, 
+       uint64_t uid)
+{
+  unsigned long long *need = cls;
+
+  if (NULL == key)
+    {
+      GNUNET_free (need);
+      return GNUNET_SYSERR;
+    }
+  if (size + GNUNET_DATASTORE_ENTRY_OVERHEAD > *need)
+    *need = 0;
+  else
+    *need -= size + GNUNET_DATASTORE_ENTRY_OVERHEAD;
+  plugin->api->next_request (next_cls, 
+                            (0 == *need) ? GNUNET_YES : GNUNET_NO);
+#if DEBUG_DATASTORE
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+             "Deleting %llu bytes of low-priority content (still trying to free another %llu bytes)\n",
+             size + GNUNET_DATASTORE_ENTRY_OVERHEAD,
+             *need);
+#endif
+  GNUNET_STATISTICS_update (stats,
+                           gettext_noop ("# bytes purged (low-priority)"),
+                           size,
+                           GNUNET_NO);
+  GNUNET_CONTAINER_bloomfilter_remove (filter,
+                                      key);
+  return GNUNET_NO;
+}
+
+
+/**
+ * Manage available disk space by running tasks
+ * that will discard content if necessary.  This
+ * function will be run whenever a request for
+ * "need" bytes of storage could only be satisfied
+ * by eating into the "cache" (and we want our cache
+ * space back).
+ *
+ * @param need number of bytes of content that were
+ *        placed into the "cache" (and hence the
+ *        number of bytes that should be removed).
+ */
+static void
+manage_space (unsigned long long need)
+{
+  unsigned long long *n;
+
+#if DEBUG_DATASTORE
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+             "Asked to free up %llu bytes of cache space\n",
+             need);
+#endif
+  n = GNUNET_malloc (sizeof(unsigned long long));
+  *n = need;
+  plugin->api->iter_low_priority (plugin->api->cls,
+                                 0,
+                                 &manage,
+                                 n);
+}
+
 
 /**
  * Function called to notify a client about the socket
@@ -198,6 +460,10 @@ transmit_callback (void *cls,
   struct TransmitCallbackContext *tcc = cls;
   size_t msize;
   
+  tcc->th = NULL;
+  GNUNET_CONTAINER_DLL_remove (tcc_head,
+                              tcc_tail,
+                              tcc);
   msize = ntohs(tcc->msg->size);
   if (size == 0)
     {
@@ -208,9 +474,8 @@ transmit_callback (void *cls,
       if (tcc->tc != NULL)
        tcc->tc (tcc->tc_cls, GNUNET_SYSERR);
       if (GNUNET_YES == tcc->end)
-       {
-         GNUNET_SERVER_receive_done (tcc->client, GNUNET_SYSERR);
-       }
+       GNUNET_SERVER_receive_done (tcc->client, GNUNET_SYSERR);       
+      GNUNET_SERVER_client_drop (tcc->client);
       GNUNET_free (tcc->msg);
       GNUNET_free (tcc);
       return 0;
@@ -221,10 +486,6 @@ transmit_callback (void *cls,
     tcc->tc (tcc->tc_cls, GNUNET_OK);
   if (GNUNET_YES == tcc->end)
     {
-#if DEBUG_DATASTORE
-      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-                 "Request completed, ready for the next request!\n");
-#endif
       GNUNET_SERVER_receive_done (tcc->client, GNUNET_OK);
     }
   else
@@ -234,6 +495,7 @@ transmit_callback (void *cls,
                  "Response transmitted, more pending!\n");
 #endif
     }
+  GNUNET_SERVER_client_drop (tcc->client);
   GNUNET_free (tcc->msg);
   GNUNET_free (tcc);
   return msize;
@@ -245,6 +507,8 @@ transmit_callback (void *cls,
  *
  * @param client target of the message
  * @param msg message to transmit, will be freed!
+ * @param tc function to call afterwards
+ * @param tc_cls closure for tc
  * @param end is this the last response (and we should
  *        signal the server completion accodingly after
  *        transmitting this message)?
@@ -258,19 +522,24 @@ transmit (struct GNUNET_SERVER_Client *client,
 {
   struct TransmitCallbackContext *tcc;
 
+  if (GNUNET_YES == cleaning_done)
+    {
+      if (NULL != tc)
+       tc (tc_cls, GNUNET_SYSERR);
+      return;
+    }
   tcc = GNUNET_malloc (sizeof(struct TransmitCallbackContext));
   tcc->msg = msg;
   tcc->client = client;
   tcc->tc = tc;
   tcc->tc_cls = tc_cls;
   tcc->end = end;
-
   if (NULL ==
-      GNUNET_SERVER_notify_transmit_ready (client,
-                                          ntohs(msg->size),
-                                          GNUNET_TIME_UNIT_FOREVER_REL,
-                                          &transmit_callback,
-                                          tcc))
+      (tcc->th = GNUNET_SERVER_notify_transmit_ready (client,
+                                                     ntohs(msg->size),
+                                                     GNUNET_TIME_UNIT_FOREVER_REL,
+                                                     &transmit_callback,
+                                                     tcc)))
     {
       GNUNET_break (0);
       if (GNUNET_YES == end)
@@ -285,7 +554,12 @@ transmit (struct GNUNET_SERVER_Client *client,
        tc (tc_cls, GNUNET_SYSERR);
       GNUNET_free (msg);
       GNUNET_free (tcc);
+      return;
     }
+  GNUNET_SERVER_client_keep (client);
+  GNUNET_CONTAINER_DLL_insert (tcc_head,
+                              tcc_tail,
+                              tcc);
 }
 
 
@@ -306,7 +580,7 @@ transmit_status (struct GNUNET_SERVER_Client *client,
 
 #if DEBUG_DATASTORE
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-             "Transmitting `%s' message with value %d and message %s\n",
+             "Transmitting `%s' message with value %d and message `%s'\n",
              "STATUS",
              code,
              msg != NULL ? msg : "(none)");
@@ -316,7 +590,8 @@ transmit_status (struct GNUNET_SERVER_Client *client,
   sm->header.size = htons(sizeof(struct StatusMessage) + slen);
   sm->header.type = htons(GNUNET_MESSAGE_TYPE_DATASTORE_STATUS);
   sm->status = htonl(code);
-  memcpy (&sm[1], msg, slen);  
+  if (slen > 0)
+    memcpy (&sm[1], msg, slen);  
   transmit (client, &sm->header, NULL, NULL, GNUNET_YES);
 }
 
@@ -325,7 +600,7 @@ transmit_status (struct GNUNET_SERVER_Client *client,
  * Function called once the transmit operation has
  * either failed or succeeded.
  *
- * @param cls closure
+ * @param next_cls closure for calling "next_request" callback
  * @param status GNUNET_OK on success, GNUNET_SYSERR on error
  */
 static void 
@@ -335,8 +610,9 @@ get_next(void *next_cls,
   if (status != GNUNET_OK)
     {
       GNUNET_log (GNUNET_ERROR_TYPE_INFO,
-                 _("Failed to transmit an item to the client; aborting iteration.\n"));    
-      plugin->api->next_request (next_cls, GNUNET_YES);
+                 _("Failed to transmit an item to the client; aborting iteration.\n"));
+      if (plugin != NULL)
+       plugin->api->next_request (next_cls, GNUNET_YES);
       return;
     }
   plugin->api->next_request (next_cls, GNUNET_NO);
@@ -410,6 +686,10 @@ transmit_item (void *cls,
              "Transmitting `%s' message\n",
              "DATA");
 #endif
+  GNUNET_STATISTICS_update (stats,
+                           gettext_noop ("# results found"),
+                           1,
+                           GNUNET_NO);
   transmit (client, &dm->header, &get_next, next_cls, GNUNET_NO);
   return GNUNET_OK;
 }
@@ -432,27 +712,46 @@ handle_reserve (void *cls,
   unsigned long long used;
   unsigned long long req;
   uint64_t amount;
-  uint64_t entries;
+  uint32_t entries;
 
 #if DEBUG_DATASTORE
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Processing `%s' request\n",
              "RESERVE");
 #endif
-  amount = ntohl(msg->amount);
-  entries = GNUNET_ntohll(msg->entries);
+  amount = GNUNET_ntohll(msg->amount);
+  entries = ntohl(msg->entries);
   used = plugin->api->get_size (plugin->api->cls) + reserved;
   req = amount + ((unsigned long long) GNUNET_DATASTORE_ENTRY_OVERHEAD) * entries;
   if (used + req > quota)
     {
       if (quota < used)
-       used = quota; /* cheat a bit */
+       used = quota; /* cheat a bit for error message (to avoid negative numbers) */
       GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
                  _("Insufficient space (%llu bytes are available) to satisfy `%s' request for %llu bytes\n"),
                  quota - used,
                  "RESERVE",
                  req);
-      transmit_status (client, 0, gettext_noop ("Insufficient space to satisfy request"));
+      if (cache_size < req)
+       {
+         /* TODO: document this in the FAQ; essentially, if this
+            message happens, the insertion request could be blocked
+            by less-important content from migration because it is
+            larger than 1/8th of the overall available space, and
+            we only reserve 1/8th for "fresh" insertions */
+         GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+                     _("The requested amount (%llu bytes) is larger than the cache size (%llu bytes)\n"),
+                     req,
+                     cache_size);
+         transmit_status (client, 0, 
+                          gettext_noop ("Insufficient space to satisfy request and "
+                                        "requested amount is larger than cache size"));
+       }
+      else
+       {
+         transmit_status (client, 0, 
+                          gettext_noop ("Insufficient space to satisfy request"));
+       }
       return;      
     }
   reserved += req;
@@ -517,7 +816,6 @@ handle_release_reserve (void *cls,
          return;
        }       
       prev = pos;
-      pos = next;
     }
   GNUNET_break (0);
   transmit_status (client, GNUNET_SYSERR, gettext_noop ("Could not find matching reservation"));
@@ -617,12 +915,26 @@ handle_put (void *cls,
                          GNUNET_TIME_absolute_ntoh(dm->expiration),
                          &msg);
   if (GNUNET_OK == ret)
-    GNUNET_CONTAINER_bloomfilter_add (filter,
-                                     &dm->key);
+    {
+      GNUNET_STATISTICS_update (stats,
+                               gettext_noop ("# bytes stored"),
+                               size,
+                               GNUNET_NO);
+      GNUNET_CONTAINER_bloomfilter_add (filter,
+                                       &dm->key);
+#if DEBUG_DATASTORE
+      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                 "Successfully stored %u bytes under key `%s'\n",
+                 size,
+                 GNUNET_h2s (&dm->key));
+#endif
+    }
   transmit_status (client, 
                   (GNUNET_SYSERR == ret) ? GNUNET_SYSERR : GNUNET_OK, 
                   msg);
   GNUNET_free_non_null (msg);
+  if (quota - reserved - cache_size < plugin->api->get_size (plugin->api->cls))
+    manage_space (size + GNUNET_DATASTORE_ENTRY_OVERHEAD);
 }
 
 
@@ -638,7 +950,6 @@ handle_get (void *cls,
             struct GNUNET_SERVER_Client *client,
             const struct GNUNET_MessageHeader *message)
 {
-  static struct GNUNET_TIME_Absolute zero;
   const struct GetMessage *msg;
   uint16_t size;
 
@@ -655,6 +966,11 @@ handle_get (void *cls,
       GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
       return;
     }
+  GNUNET_STATISTICS_update (stats,
+                           gettext_noop ("# GET requests received"),
+                           1,
+                           GNUNET_NO);
+  GNUNET_SERVER_client_keep (client);
   msg = (const struct GetMessage*) message;
   if ( (size == sizeof(struct GetMessage)) &&
        (GNUNET_YES != GNUNET_CONTAINER_bloomfilter_test (filter,
@@ -663,14 +979,19 @@ handle_get (void *cls,
       /* don't bother database... */
 #if DEBUG_DATASTORE
       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-                 "Empty result set for `%s' request.\n",
-                 "GET");
+                 "Empty result set for `%s' request for `%s'.\n",
+                 "GET",
+                 GNUNET_h2s (&msg->key));
 #endif 
+      GNUNET_STATISTICS_update (stats,
+                               gettext_noop ("# requests filtered by bloomfilter"),
+                               1,
+                               GNUNET_NO);
       transmit_item (client,
-                    NULL, NULL, 0, NULL, 0, 0, 0, zero, 0);
+                    NULL, NULL, 0, NULL, 0, 0, 0, 
+                    GNUNET_TIME_UNIT_ZERO_ABS, 0);
       return;
     }
-  GNUNET_SERVER_client_keep (client);
   plugin->api->get (plugin->api->cls,
                    ((size == sizeof(struct GetMessage)) ? &msg->key : NULL),
                    NULL,
@@ -701,6 +1022,10 @@ handle_update (void *cls,
              "Processing `%s' request\n",
              "UPDATE");
 #endif
+  GNUNET_STATISTICS_update (stats,
+                           gettext_noop ("# UPDATE requests received"),
+                           1,
+                           GNUNET_NO);
   msg = (const struct UpdateMessage*) message;
   emsg = NULL;
   ret = plugin->api->update (plugin->api->cls,
@@ -730,6 +1055,10 @@ handle_get_random (void *cls,
              "Processing `%s' request\n",
              "GET_RANDOM");
 #endif
+  GNUNET_STATISTICS_update (stats,
+                           gettext_noop ("# GET RANDOM requests received"),
+                           1,
+                           GNUNET_NO);
   GNUNET_SERVER_client_keep (client);
   plugin->api->iter_migration_order (plugin->api->cls,
                                     0,
@@ -783,7 +1112,7 @@ remove_callback (void *cls,
       if (GNUNET_YES == rc->found)
        transmit_status (rc->client, GNUNET_OK, NULL);       
       else
-       transmit_status (rc->client, GNUNET_SYSERR, _("Content not found"));            
+       transmit_status (rc->client, GNUNET_NO, _("Content not found"));        
       GNUNET_SERVER_client_drop (rc->client);
       GNUNET_free (rc);
       return GNUNET_OK; /* last item */
@@ -795,6 +1124,10 @@ remove_callback (void *cls,
              (unsigned long long) uid,
              "REMOVE");
 #endif 
+  GNUNET_STATISTICS_update (stats,
+                           gettext_noop ("# bytes removed (explicit request)"),
+                           size,
+                           GNUNET_NO);
   GNUNET_CONTAINER_bloomfilter_remove (filter,
                                       key);
   plugin->api->next_request (next_cls, GNUNET_YES);
@@ -829,13 +1162,16 @@ handle_remove (void *cls,
       GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
       return;
     }
+  GNUNET_STATISTICS_update (stats,
+                           gettext_noop ("# REMOVE requests received"),
+                           1,
+                           GNUNET_NO);
   rc = GNUNET_malloc (sizeof(struct RemoveContext));
   GNUNET_SERVER_client_keep (client);
   rc->client = client;
   GNUNET_CRYPTO_hash (&dm[1],
                      ntohl(dm->size),
                      &vhash);
-  GNUNET_SERVER_client_keep (client);
   plugin->api->get (plugin->api->cls,
                    &dm->key,
                    &vhash,
@@ -894,8 +1230,7 @@ static struct GNUNET_SERVER_MessageHandler handlers[] = {
  * Load the datastore plugin.
  */
 static struct DatastorePlugin *
-load_plugin (struct GNUNET_CONFIGURATION_Handle *cfg,
-            struct GNUNET_SCHEDULER_Handle *sched)
+load_plugin () 
 {
   struct DatastorePlugin *ret;
   char *libname;
@@ -953,6 +1288,29 @@ unload_plugin (struct DatastorePlugin *plug)
 }
 
 
+/**
+ * Final task run after shutdown.  Unloads plugins and disconnects us from
+ * statistics.
+ */
+static void
+unload_task (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
+{
+  unload_plugin (plugin);
+  plugin = NULL;
+  if (filter != NULL)
+    {
+      GNUNET_CONTAINER_bloomfilter_free (filter);
+      filter = NULL;
+    }
+  GNUNET_ARM_stop_services (cfg, tc->sched, "statistics", NULL);
+  if (stats != NULL)
+    {
+      GNUNET_STATISTICS_destroy (stats, GNUNET_YES);
+      stats = NULL;
+    }
+}
+
+
 /**
  * Last task run during shutdown.  Disconnects us from
  * the transport and core.
@@ -960,8 +1318,34 @@ unload_plugin (struct DatastorePlugin *plug)
 static void
 cleaning_task (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
 {
-  unload_plugin (plugin);
-  plugin = NULL;
+  struct TransmitCallbackContext *tcc;
+
+  cleaning_done = GNUNET_YES;
+  while (NULL != (tcc = tcc_head))
+    {
+      GNUNET_CONTAINER_DLL_remove (tcc_head,
+                                  tcc_tail,
+                                  tcc);
+      if (tcc->th != NULL)
+       {
+         GNUNET_CONNECTION_notify_transmit_ready_cancel (tcc->th);
+         GNUNET_SERVER_client_drop (tcc->client);
+       }
+   if (NULL != tcc->tc)
+       tcc->tc (tcc->tc_cls, GNUNET_SYSERR);
+      GNUNET_free (tcc->msg);
+      GNUNET_free (tcc);
+    }
+  if (expired_kill_task != GNUNET_SCHEDULER_NO_TASK)
+    {
+      GNUNET_SCHEDULER_cancel (sched,
+                              expired_kill_task);
+      expired_kill_task = GNUNET_SCHEDULER_NO_TASK;
+    }
+  GNUNET_SCHEDULER_add_continuation (sched,
+                                    &unload_task,
+                                    NULL,
+                                    GNUNET_SCHEDULER_REASON_PREREQ_DONE);
 }
 
 
@@ -978,7 +1362,32 @@ cleanup_reservations (void *cls,
                      struct GNUNET_SERVER_Client
                      * client)
 {
-  /* FIXME */
+  struct ReservationList *pos;
+  struct ReservationList *prev;
+  struct ReservationList *next;
+
+  if (client == NULL)
+    return;
+  prev = NULL;
+  pos = reservations;
+  while (NULL != pos)
+    {
+      next = pos->next;
+      if (pos->client == client)
+       {
+         if (prev == NULL)
+           reservations = next;
+         else
+           prev->next = next;
+         reserved -= pos->amount + pos->entries * GNUNET_DATASTORE_ENTRY_OVERHEAD;
+         GNUNET_free (pos);
+       }
+      else
+       {
+         prev = pos;
+       }
+      pos = next;
+    }
 }
 
 
@@ -986,19 +1395,21 @@ cleanup_reservations (void *cls,
  * Process datastore requests.
  *
  * @param cls closure
- * @param sched scheduler to use
+ * @param s scheduler to use
  * @param server the initialized server
- * @param cfg configuration to use
+ * @param c configuration to use
  */
 static void
 run (void *cls,
-     struct GNUNET_SCHEDULER_Handle *sched,
+     struct GNUNET_SCHEDULER_Handle *s,
      struct GNUNET_SERVER_Handle *server,
-     struct GNUNET_CONFIGURATION_Handle *cfg)
+     const struct GNUNET_CONFIGURATION_Handle *c)
 {
   char *fn;
   unsigned int bf_size;
 
+  sched = s;
+  cfg = c;
   if (GNUNET_OK !=
       GNUNET_CONFIGURATION_get_value_number (cfg,
                                              "DATASTORE", "QUOTA", &quota))
@@ -1009,6 +1420,8 @@ run (void *cls,
                  "DATASTORE");
       return;
     }
+  stats = GNUNET_STATISTICS_create (sched, "datastore", cfg);
+  cache_size = quota / 8; /* Or should we make this an option? */
   bf_size = quota / 32; /* 8 bit per entry, 1 bit per 32 kb in DB */
   fn = NULL;
   if ( (GNUNET_OK !=
@@ -1031,22 +1444,37 @@ run (void *cls,
     {
       GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
                  _("Failed to initialize bloomfilter.\n"));
+      if (stats != NULL)
+       {
+         GNUNET_STATISTICS_destroy (stats, GNUNET_YES);
+         stats = NULL;
+       }
       return;
     }
-  plugin = load_plugin (cfg, sched);
+  GNUNET_ARM_start_services (cfg, sched, "statistics", NULL);
+  plugin = load_plugin ();
   if (NULL == plugin)
     {
       GNUNET_CONTAINER_bloomfilter_free (filter);
+      filter = NULL;
+      GNUNET_ARM_stop_services (cfg, sched, "statistics", NULL);
+      if (stats != NULL)
+       {
+         GNUNET_STATISTICS_destroy (stats, GNUNET_YES);
+         stats = NULL;
+       }
       return;
     }
   GNUNET_SERVER_disconnect_notify (server, &cleanup_reservations, NULL);
   GNUNET_SERVER_add_handlers (server, handlers);
+  expired_kill_task
+    = GNUNET_SCHEDULER_add_with_priority (sched,
+                                         GNUNET_SCHEDULER_PRIORITY_IDLE,
+                                         &delete_expired, NULL);
   GNUNET_SCHEDULER_add_delayed (sched,
-                                GNUNET_YES,
-                                GNUNET_SCHEDULER_PRIORITY_IDLE,
-                                GNUNET_SCHEDULER_NO_PREREQUISITE_TASK,
                                 GNUNET_TIME_UNIT_FOREVER_REL,
                                 &cleaning_task, NULL);
+  
 }
 
 
@@ -1065,7 +1493,9 @@ main (int argc, char *const *argv)
   ret = (GNUNET_OK ==
          GNUNET_SERVICE_run (argc,
                              argv,
-                             "datastore", &run, NULL, NULL, NULL)) ? 0 : 1;
+                             "datastore",
+                            GNUNET_SERVICE_OPTION_NONE,
+                            &run, NULL)) ? 0 : 1;
   return ret;
 }