fixes

7 files changed, 141 insertions, 73 deletions

diff --git a/src/fs/Makefile.am b/src/fs/Makefile.am
index 9c9fd7525..6c850a91d 100644
--- a/src/fs/Makefile.am
+++ b/src/fs/Makefile.am
@@ -94,7 +94,6 @@ gnunet_search_LDADD = \
 
 gnunet_service_fs_SOURCES =  \
  gnunet-service-fs.c \
- gnunet-service-fs_drq.c gnunet-service-fs_drq.h \
  gnunet-service-fs_indexing.c gnunet-service-fs_indexing.h 
 gnunet_service_fs_LDADD =  \
  $(top_builddir)/src/fs/libgnunetfs.la \
diff --git a/src/fs/fs_namespace.c b/src/fs/fs_namespace.c
index 9e198adc2..e8b97bb19 100644
--- a/src/fs/fs_namespace.c
+++ b/src/fs/fs_namespace.c
@@ -79,6 +79,11 @@ struct AdvertisementContext
   struct GNUNET_DATASTORE_Handle *dsh;
 
   /**
+   * Our scheduler.
+   */
+  struct GNUNET_SCHEDULER_Handle *sched;
+
+  /**
    * Our KSK URI.
    */ 
   struct GNUNET_FS_Uri *ksk_uri;
@@ -126,6 +131,23 @@ struct AdvertisementContext
 
 
 /**
+ * Disconnect from the datastore.
+ * 
+ * @param cls datastore handle
+ * @param tc scheduler context
+ */
+static void
+do_disconnect (void *cls,
+               const struct GNUNET_SCHEDULER_TaskContext *tc)
+{
+  struct GNUNET_DATASTORE_Handle *dsh = cls;
+
+  GNUNET_DATASTORE_disconnect (dsh, 
+                               GNUNET_NO);
+}
+
+
+/**
  * Continuation called to notify client about result of the
  * operation.
  *
@@ -149,7 +171,10 @@ advertisement_cont (void *cls,
   if (GNUNET_OK != success)
     {
       /* error! */
-      GNUNET_DATASTORE_disconnect (ac->dsh, GNUNET_NO);
+      GNUNET_SCHEDULER_add_continuation (ac->sched,
+                                         &do_disconnect,
+                                         ac->dsh,
+                                         GNUNET_SCHEDULER_REASON_PREREQ_DONE);
       ac->cont (ac->cont_cls, NULL, msg);
       GNUNET_FS_uri_destroy (ac->ksk_uri);
       GNUNET_free (ac->pt);
@@ -161,7 +186,10 @@ advertisement_cont (void *cls,
   if (ac->pos == ac->ksk_uri->data.ksk.keywordCount)
     {
       /* done! */
-      GNUNET_DATASTORE_disconnect (ac->dsh, GNUNET_NO);
+      GNUNET_SCHEDULER_add_continuation (ac->sched,
+                                         &do_disconnect,
+                                         ac->dsh,
+                                         GNUNET_SCHEDULER_REASON_PREREQ_DONE);
       ac->cont (ac->cont_cls, ac->ksk_uri, NULL);
       GNUNET_FS_uri_destroy (ac->ksk_uri);
       GNUNET_free (ac->pt);
@@ -296,6 +324,7 @@ GNUNET_FS_namespace_advertise (struct GNUNET_FS_Handle *h,
   ctx->cont = cont;
   ctx->cont_cls = cont_cls;
   ctx->dsh = dsh;
+  ctx->sched = h->sched;
   ctx->ksk_uri = GNUNET_FS_uri_dup (ksk_uri);
   ctx->nb = nb;
   ctx->pt = pt;
diff --git a/src/fs/fs_publish.c b/src/fs/fs_publish.c
index aa7f794be..4e2039ba0 100644
--- a/src/fs/fs_publish.c
+++ b/src/fs/fs_publish.c
@@ -109,11 +109,14 @@ GNUNET_FS_publish_make_status_ (struct GNUNET_FS_ProgressInfo *pi,
 /**
  * Cleanup the publish context, we're done with it.
  *
- * @param pc struct to clean up after
+ * @param cls struct to clean up after
+ * @param tc scheduler context
  */
 static void
-publish_cleanup (struct GNUNET_FS_PublishContext *pc)
+publish_cleanup (void *cls,
+                 const struct GNUNET_SCHEDULER_TaskContext *tc)
 {
+  struct GNUNET_FS_PublishContext *pc = cls;
   GNUNET_FS_file_information_destroy (pc->fi, NULL, NULL);
   if (pc->namespace != NULL)
     GNUNET_FS_namespace_delete (pc->namespace, GNUNET_NO);
@@ -146,9 +149,11 @@ ds_put_cont (void *cls,
 
   if (GNUNET_SYSERR == pcc->sc->in_network_wait)
     {
-      /* we were aborted in the meantime,
-         finish shutdown! */
-      publish_cleanup (pcc->sc);
+      /* we were aborted in the meantime, finish shutdown! */
+      GNUNET_SCHEDULER_add_continuation (pcc->sc->h->sched,                                      
+                                         &publish_cleanup,
+                                         pcc->sc,
+                                         GNUNET_SCHEDULER_REASON_PREREQ_DONE);
       GNUNET_free (pcc);
       return;
     }
@@ -1016,7 +1021,10 @@ GNUNET_FS_publish_signal_suspend_ (void *cls)
                                       &fip_signal_suspend,
                                       pc);
   GNUNET_FS_end_top (pc->h, pc->top);
-  publish_cleanup (pc);
+  GNUNET_SCHEDULER_add_continuation (pc->h->sched,                                       
+                                     &publish_cleanup,
+                                     pc,
+                                     GNUNET_SCHEDULER_REASON_PREREQ_DONE);
 }
 
 /**
@@ -1163,7 +1171,10 @@ GNUNET_FS_publish_stop (struct GNUNET_FS_PublishContext *pc)
       pc->in_network_wait = GNUNET_SYSERR;
       return;
     }
-  publish_cleanup (pc);
+  GNUNET_SCHEDULER_add_continuation (pc->h->sched,                                       
+                                     &publish_cleanup,
+                                     pc,
+                                     GNUNET_SCHEDULER_REASON_PREREQ_DONE);
 }
 
 
diff --git a/src/fs/gnunet-service-fs.c b/src/fs/gnunet-service-fs.c
index 7e88f9d4e..e85d99cfd 100644
--- a/src/fs/gnunet-service-fs.c
+++ b/src/fs/gnunet-service-fs.c
@@ -43,7 +43,6 @@
 #include "gnunet_signatures.h"
 #include "gnunet_statistics_service.h"
 #include "gnunet_util_lib.h"
-#include "gnunet-service-fs_drq.h"
 #include "gnunet-service-fs_indexing.h"
 #include "fs.h"
 
@@ -89,6 +88,11 @@ 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.
@@ -472,10 +476,10 @@ struct PendingRequest
   GNUNET_PEER_Id *used_pids;
   
   /**
-   * Our entry in the DRQ (non-NULL while we wait for our
+   * Our entry in the queue (non-NULL while we wait for our
    * turn to interact with the local database).
    */
-  struct DatastoreRequestQueue *drq;
+  struct GNUNET_DATASTORE_QueueEntry *qe;
 
   /**
    * Size of the 'bf' (in bytes).
@@ -696,10 +700,10 @@ destroy_pending_request (struct PendingRequest *pr)
   (void) GNUNET_CONTAINER_multihashmap_remove (query_request_map,
                                                &pr->query,
                                                pr);
-  if (pr->drq != NULL)
-    {
-      GNUNET_FS_drq_get_cancel (pr->drq);
-      pr->drq = NULL;
+  if (pr->qe != NULL)
+     {
+      GNUNET_DATASTORE_cancel (pr->qe);
+      pr->qe = NULL;
     }
   if (pr->client_request_list != NULL)
     {
@@ -995,6 +999,9 @@ shutdown_task (void *cls,
       GNUNET_STATISTICS_destroy (stats, GNUNET_NO);
       stats = NULL;
     }
+  GNUNET_DATASTORE_disconnect (dsh,
+                               GNUNET_NO);
+  dsh = NULL;
   sched = NULL;
   cfg = NULL;  
 }
@@ -1852,13 +1859,13 @@ process_reply (void *cls,
       /* only possible reply, stop requesting! */
       while (NULL != pr->pending_head)
         destroy_pending_message_list_entry (pr->pending_head);
-      if (pr->drq != NULL)
+      if (pr->qe != NULL)
         {
           if (pr->client_request_list != NULL)
             GNUNET_SERVER_receive_done (pr->client_request_list->client_list->client, 
                                         GNUNET_YES);
-          GNUNET_FS_drq_get_cancel (pr->drq);
-          pr->drq = NULL;
+          GNUNET_DATASTORE_cancel (pr->qe);
+          pr->qe = NULL;
         }
       pr->do_remove = GNUNET_YES;
       if (pr->task != GNUNET_SCHEDULER_NO_TASK)
@@ -2209,7 +2216,7 @@ process_local_reply (void *cls,
       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                   "Done processing local replies, forwarding request to other peers.\n");
 #endif
-      pr->drq = NULL;
+      pr->qe = NULL;
       if (pr->client_request_list != NULL)
         {
           GNUNET_SERVER_receive_done (pr->client_request_list->client_list->client, 
@@ -2263,7 +2270,7 @@ process_local_reply (void *cls,
                                             anonymity, expiration, uid, 
                                             &process_local_reply,
                                             pr))
-        GNUNET_FS_drq_get_next (GNUNET_YES);
+        GNUNET_DATASTORE_get_next (dsh, GNUNET_YES);
       return;
     }
   /* check for duplicates */
@@ -2284,7 +2291,7 @@ process_local_reply (void *cls,
                                 gettext_noop ("# results filtered by query bloomfilter"),
                                 1,
                                 GNUNET_NO);
-      GNUNET_FS_drq_get_next (GNUNET_YES);
+      GNUNET_DATASTORE_get_next (dsh, GNUNET_YES);
       return;
     }
 #if DEBUG_FS
@@ -2315,7 +2322,7 @@ process_local_reply (void *cls,
     {
       GNUNET_break (0);
       /* FIXME: consider removing the block? */
-      GNUNET_FS_drq_get_next (GNUNET_YES);
+      GNUNET_DATASTORE_get_next (dsh, GNUNET_YES);
       return;
     }
   prq.type = type;
@@ -2325,7 +2332,7 @@ process_local_reply (void *cls,
   if ( (type == GNUNET_BLOCK_TYPE_DBLOCK) ||
        (type == GNUNET_BLOCK_TYPE_IBLOCK) ) 
     {
-      GNUNET_FS_drq_get_next (GNUNET_NO);
+      GNUNET_DATASTORE_get_next (dsh, GNUNET_NO);
       return;
     }
   if ( (pr->client_request_list == NULL) &&
@@ -2340,10 +2347,10 @@ process_local_reply (void *cls,
                                 gettext_noop ("# processing result set cut short due to load"),
                                 1,
                                 GNUNET_NO);
-      GNUNET_FS_drq_get_next (GNUNET_NO);
+      GNUNET_DATASTORE_get_next (dsh, GNUNET_NO);
       return;
     }
-  GNUNET_FS_drq_get_next (GNUNET_YES);
+  GNUNET_DATASTORE_get_next (dsh, GNUNET_YES);
 }
 
 
@@ -2656,12 +2663,14 @@ handle_p2p_get (void *cls,
     type = GNUNET_BLOCK_TYPE_ANY; /* to get on-demand as well */
   timeout = GNUNET_TIME_relative_multiply (BASIC_DATASTORE_REQUEST_DELAY,
                                            (pr->priority + 1)); 
-  pr->drq = GNUNET_FS_drq_get (&gm->query,
-                               type,                           
-                               &process_local_reply,
-                               pr,
-                               timeout,
-                               GNUNET_NO);
+  pr->qe = GNUNET_DATASTORE_get (dsh,
+                                 &gm->query,
+                                 type,                         
+                                 (unsigned int) preference, 64 /* FIXME */,
+                                 
+                                 timeout,
+                                 &process_local_reply,
+                                 pr);
 
   /* Are multiple results possible?  If so, start processing remotely now! */
   switch (pr->type)
@@ -2852,12 +2861,13 @@ handle_start_search (void *cls,
                                                    GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE));
   if (type == GNUNET_BLOCK_TYPE_DBLOCK)
     type = GNUNET_BLOCK_TYPE_ANY; /* get on-demand blocks too! */
-  pr->drq = GNUNET_FS_drq_get (&sm->query,
-                               type,                           
-                               &process_local_reply,
-                               pr,
-                               GNUNET_CONSTANTS_SERVICE_TIMEOUT,
-                               GNUNET_YES);
+  pr->qe = GNUNET_DATASTORE_get (dsh,
+                                 &sm->query,
+                                 type,
+                                 -3, -1,
+                                 GNUNET_CONSTANTS_SERVICE_TIMEOUT,                             
+                                 &process_local_reply,
+                                 pr);
 }
 
 
@@ -2937,9 +2947,13 @@ main_init (struct GNUNET_SCHEDULER_Handle *s,
       requests_by_expiration_heap = NULL;
       GNUNET_CONTAINER_multihashmap_destroy (peer_request_map);
       peer_request_map = NULL;
-
+      if (dsh != NULL)
+        {
+          GNUNET_DATASTORE_disconnect (dsh, GNUNET_NO);
+          dsh = NULL;
+        }
       return GNUNET_SYSERR;
-    }  
+    }
   GNUNET_SERVER_disconnect_notify (server, 
                                    &handle_client_disconnect,
                                    NULL);
@@ -2969,11 +2983,19 @@ run (void *cls,
   active_migration = GNUNET_CONFIGURATION_get_value_yesno (cfg,
                                                            "FS",
                                                            "ACTIVEMIGRATION");
-  if ( (GNUNET_OK != GNUNET_FS_drq_init (sched, cfg)) ||
-       (GNUNET_OK != GNUNET_FS_indexing_init (sched, cfg)) ||
+  dsh = GNUNET_DATASTORE_connect (cfg,
+                                  sched);
+  if (dsh == NULL)
+    {
+      GNUNET_SCHEDULER_shutdown (sched);
+      return;
+    }
+  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;
       return;   
     }
 }
diff --git a/src/fs/gnunet-service-fs_drq.c b/src/fs/gnunet-service-fs_drq.c
index ab6c9ad21..b549c6707 100644
--- a/src/fs/gnunet-service-fs_drq.c
+++ b/src/fs/gnunet-service-fs_drq.c
@@ -113,10 +113,6 @@ static struct DatastoreRequestQueue *drq_head;
  */
 static struct DatastoreRequestQueue *drq_tail;
 
-/**
- * Our connection to the datastore.
- */
-static struct GNUNET_DATASTORE_Handle *dsh;
 
 /**
  * Pointer to the currently actively running request,
@@ -295,9 +291,6 @@ shutdown_task (void *cls,
               "DRQ shutdown initiated\n");
 #endif
   GNUNET_assert (NULL != dsh);
-  GNUNET_DATASTORE_disconnect (dsh,
-                               GNUNET_NO);
-  dsh = NULL;
   while (NULL != (drq = drq_head))
     {
       drq_head = drq->next;
@@ -465,7 +458,6 @@ drq_remove_cont (void *cls,
   rc->cont (rc->cont_cls,
             success,
             msg);
-  GNUNET_DATASTORE_disconnect (rc->rmdsh, GNUNET_NO);
   GNUNET_free (rc);
 }
 
@@ -494,8 +486,6 @@ GNUNET_FS_drq_remove (const GNUNET_HashCode *key,
   struct GNUNET_DATASTORE_Handle *rmdsh; 
   struct RemoveContext *rc;
 
-  rmdsh = GNUNET_DATASTORE_connect (cfg,
-                                    sched);
   if (rmdsh == NULL)
     {
       GNUNET_break (0);
diff --git a/src/fs/gnunet-service-fs_indexing.c b/src/fs/gnunet-service-fs_indexing.c
index f8593ffd4..363bdea41 100644
--- a/src/fs/gnunet-service-fs_indexing.c
+++ b/src/fs/gnunet-service-fs_indexing.c
@@ -101,6 +101,12 @@ static struct GNUNET_SCHEDULER_Handle *sched;
  */
 static const struct GNUNET_CONFIGURATION_Handle *cfg;
 
+/**
+ * Datastore handle.  Created and destroyed by code in
+ * gnunet-service-fs (this is an alias).
+ */
+static struct GNUNET_DATASTORE_Handle *dsh;
+
 
 /**
  * Write the current index information list to disk.
@@ -575,12 +581,14 @@ GNUNET_FS_handle_on_demand_block (const GNUNET_HashCode * key,
   if (size != sizeof (struct OnDemandBlock))
     {
       GNUNET_break (0);
-      GNUNET_FS_drq_remove (key,
-                            size,
-                            data,
-                            &remove_cont,
-                            NULL,
-                            GNUNET_TIME_UNIT_FOREVER_REL);
+      GNUNET_DATASTORE_remove (dsh,
+                               key,
+                               size,
+                               data,
+                               -1, -1,
+                               GNUNET_TIME_UNIT_FOREVER_REL,
+                               &remove_cont,
+                               NULL);
       return GNUNET_SYSERR;
     }
   odb = (const struct OnDemandBlock*) data;
@@ -608,12 +616,14 @@ GNUNET_FS_handle_on_demand_block (const GNUNET_HashCode * key,
                   STRERROR (errno));
       if (fh != NULL)
         GNUNET_DISK_file_close (fh);
-      GNUNET_FS_drq_remove (key,
-                            size,
-                            data,
-                            &remove_cont,
-                            NULL,
-                            GNUNET_TIME_UNIT_FOREVER_REL);
+      GNUNET_DATASTORE_remove (dsh,
+                               key,
+                               size,
+                               data,
+                               -1, -1,
+                               GNUNET_TIME_UNIT_FOREVER_REL,
+                               &remove_cont,
+                               NULL);
       return GNUNET_SYSERR;
     }
   GNUNET_DISK_file_close (fh);
@@ -637,12 +647,14 @@ GNUNET_FS_handle_on_demand_block (const GNUNET_HashCode * key,
                   _("Indexed file `%s' changed at offset %llu\n"),
                   fn,
                   (unsigned long long) off);
-      GNUNET_FS_drq_remove (key,
-                            size,
-                            data,
-                            &remove_cont,
-                            NULL,
-                            GNUNET_TIME_UNIT_FOREVER_REL);
+      GNUNET_DATASTORE_remove (dsh,
+                               key,
+                               size,
+                               data,
+                               -1, -1,
+                               GNUNET_TIME_UNIT_FOREVER_REL,
+                               &remove_cont,
+                               NULL);
       return GNUNET_SYSERR;
     }
 #if DEBUG_FS
@@ -692,13 +704,16 @@ shutdown_task (void *cls,
  *
  * @param s scheduler to use
  * @param c configuration to use
+ * @param d datastore to use
  */
 int
 GNUNET_FS_indexing_init (struct GNUNET_SCHEDULER_Handle *s,
-                         const struct GNUNET_CONFIGURATION_Handle *c)
+                         const struct GNUNET_CONFIGURATION_Handle *c,
+                         struct GNUNET_DATASTORE_Handle *d)
 {
   sched = s;
   cfg = c;
+  dsh = d;
   ifm = GNUNET_CONTAINER_multihashmap_create (128);
   GNUNET_SCHEDULER_add_delayed (sched,
                                 GNUNET_TIME_UNIT_FOREVER_REL,
diff --git a/src/fs/gnunet-service-fs_indexing.h b/src/fs/gnunet-service-fs_indexing.h
index 37df50042..489e9d3f9 100644
--- a/src/fs/gnunet-service-fs_indexing.h
+++ b/src/fs/gnunet-service-fs_indexing.h
@@ -112,11 +112,13 @@ GNUNET_FS_handle_unindex (void *cls,
  *
  * @param s scheduler to use
  * @param c configuration to use
+ * @param d datastore to use
  * @return GNUNET_OK on success
  */
 int
 GNUNET_FS_indexing_init (struct GNUNET_SCHEDULER_Handle *s,
-                         const struct GNUNET_CONFIGURATION_Handle *c);
+                         const struct GNUNET_CONFIGURATION_Handle *c,
+                         struct GNUNET_DATASTORE_Handle *d);
 
 
 #endif