[GNUnet-SVN] r25123 - gnunet/src/fs

[gnunet]

Author: grothoff
Date: 2012-11-25 00:45:27 +0100 (Sun, 25 Nov 2012)
New Revision: 25123

Modified:
   gnunet/src/fs/gnunet-service-fs.c
   gnunet/src/fs/gnunet-service-fs_pr.c
   gnunet/src/fs/gnunet-service-fs_pr.h
   gnunet/src/fs/gnunet-service-fs_put.c
Log:
actually using stream for dblock/iblock transfer if possible

Modified: gnunet/src/fs/gnunet-service-fs.c
===================================================================
--- gnunet/src/fs/gnunet-service-fs.c   2012-11-24 23:25:49 UTC (rev 25122)
+++ gnunet/src/fs/gnunet-service-fs.c   2012-11-24 23:45:27 UTC (rev 25123)
@@ -22,9 +22,6 @@
  * @file fs/gnunet-service-fs.c
  * @brief gnunet anonymity protocol implementation
  * @author Christian Grothoff
- *
- * To use:
- * - consider re-issue GSF_dht_lookup_ after non-DHT reply received
  */
 #include "platform.h"
 #include <float.h>
@@ -397,7 +394,26 @@
     GSF_pending_request_cancel_ (pr, GNUNET_YES);
     return;
   }
-  GSF_dht_lookup_ (pr);
+  if (0 == prd->anonymity_level)
+  {
+    switch (prd->type)
+    {
+    case GNUNET_BLOCK_TYPE_FS_DBLOCK:
+    case GNUNET_BLOCK_TYPE_FS_IBLOCK:
+      /* the above block types MAY be available via 'stream' */
+      GSF_stream_lookup_ (pr);
+      break; 
+    case GNUNET_BLOCK_TYPE_FS_KBLOCK:
+    case GNUNET_BLOCK_TYPE_FS_SBLOCK:
+    case GNUNET_BLOCK_TYPE_FS_NBLOCK:
+      /* the above block types are in the DHT */
+      GSF_dht_lookup_ (pr);
+      break;
+    default:
+      GNUNET_break (0);
+      break;
+    }
+  }
   consider_forwarding (NULL, pr, result);
 }
 

Modified: gnunet/src/fs/gnunet-service-fs_pr.c
===================================================================
--- gnunet/src/fs/gnunet-service-fs_pr.c        2012-11-24 23:25:49 UTC (rev 
25122)
+++ gnunet/src/fs/gnunet-service-fs_pr.c        2012-11-24 23:45:27 UTC (rev 
25123)
@@ -30,8 +30,15 @@
 #include "gnunet-service-fs_indexing.h"
 #include "gnunet-service-fs_pe.h"
 #include "gnunet-service-fs_pr.h"
+#include "gnunet-service-fs_stream.h"
 
+
 /**
+ * Desired replication level for GETs.
+ */
+#define DHT_GET_REPLICATION 5
+
+/**
  * Maximum size of the datastore queue for P2P operations.  Needs to
  * be large enough to queue MAX_QUEUE_PER_PEER operations for roughly
  * the number of active (connected) peers.
@@ -102,6 +109,11 @@
   struct GNUNET_DHT_GetHandle *gh;
 
   /**
+   * Stream request handle for this request (or NULL for none).
+   */
+  struct GSF_StreamRequest *stream_request;
+
+  /**
    * Function to call upon completion of the local get
    * request, or NULL for none.
    */
@@ -624,6 +636,11 @@
     GNUNET_DHT_get_stop (pr->gh);
     pr->gh = NULL;
   }
+  if (NULL != pr->stream_request)
+  {
+    GSF_stream_query_cancel (pr->stream_request);
+    pr->stream_request = NULL;
+  }
   if (GNUNET_SCHEDULER_NO_TASK != pr->warn_task)
   {
     GNUNET_SCHEDULER_cancel (pr->warn_task);
@@ -676,6 +693,11 @@
       GNUNET_DHT_get_stop (pr->gh);
       pr->gh = NULL;
     }
+    if (NULL != pr->stream_request)
+    {
+      GSF_stream_query_cancel (pr->stream_request);
+      pr->stream_request = NULL;
+    }
     if (GNUNET_SCHEDULER_NO_TASK != pr->warn_task)
     {
       GNUNET_SCHEDULER_cancel (pr->warn_task);
@@ -1121,7 +1143,7 @@
   pr->gh =
       GNUNET_DHT_get_start (GSF_dht, 
                             pr->public_data.type, &pr->public_data.query,
-                            5 /* DEFAULT_GET_REPLICATION */ ,
+                            DHT_GET_REPLICATION,
                             GNUNET_DHT_RO_DEMULTIPLEX_EVERYWHERE,
                             xquery, xquery_size, &handle_dht_reply, pr);
   if ( (NULL != pr->gh) && 
@@ -1133,6 +1155,72 @@
 
 
 /**
+ * Function called with a reply from the stream.
+ * 
+ * @param cls the pending request struct
+ * @param type type of the block, ANY on error
+ * @param expiration expiration time for the block
+ * @param data_size number of bytes in 'data', 0 on error
+ * @param data reply block data, NULL on error
+ */
+static void
+stream_reply_proc (void *cls,
+                  enum GNUNET_BLOCK_Type type,
+                  struct GNUNET_TIME_Absolute expiration,
+                  size_t data_size,
+                  const void *data)
+{
+  struct GSF_PendingRequest *pr = cls;
+  struct ProcessReplyClosure prq;
+  struct GNUNET_HashCode query;
+
+  pr->stream_request = NULL;
+  if (GNUNET_YES !=
+      GNUNET_BLOCK_get_key (GSF_block_ctx,
+                           type,
+                           data, data_size, &query))
+  {
+    GNUNET_break_op (0);
+    return;
+  }
+  GNUNET_STATISTICS_update (GSF_stats,
+                            gettext_noop ("# Replies received from STREAM"), 1,
+                            GNUNET_NO);
+  memset (&prq, 0, sizeof (prq));
+  prq.data = data;
+  prq.expiration = expiration;
+  /* do not allow migrated content to live longer than 1 year */
+  prq.expiration = GNUNET_TIME_absolute_min (GNUNET_TIME_relative_to_absolute 
(GNUNET_TIME_UNIT_YEARS),
+                                            prq.expiration);
+  prq.size = data_size;
+  prq.type = type;
+  process_reply (&prq, &query, pr);
+}
+
+
+/**
+ * Consider downloading via stream (if possible)
+ *
+ * @param pr the pending request to process
+ */
+void
+GSF_stream_lookup_ (struct GSF_PendingRequest *pr)
+{
+  if (0 != pr->public_data.anonymity_level)
+    return;
+  if (0 == pr->public_data.target)
+    return;
+  if (NULL != pr->stream_request)
+    return;
+  pr->stream_request = GSF_stream_query (pr->public_data.target,
+                                        &pr->public_data.query,
+                                        pr->public_data.type,
+                                        &stream_reply_proc,
+                                        pr);
+}
+
+
+/**
  * Task that issues a warning if the datastore lookup takes too long.
  *
  * @param cls the 'struct GSF_PendingRequest'
@@ -1456,6 +1544,7 @@
                    GSF_LocalLookupContinuation cont, void *cont_cls)
 {
   GNUNET_assert (NULL == pr->gh);
+  GNUNET_assert (NULL == pr->stream_request);
   GNUNET_assert (NULL == pr->llc_cont);
   pr->llc_cont = cont;
   pr->llc_cont_cls = cont_cls;

Modified: gnunet/src/fs/gnunet-service-fs_pr.h
===================================================================
--- gnunet/src/fs/gnunet-service-fs_pr.h        2012-11-24 23:25:49 UTC (rev 
25122)
+++ gnunet/src/fs/gnunet-service-fs_pr.h        2012-11-24 23:45:27 UTC (rev 
25123)
@@ -93,8 +93,7 @@
   const struct GNUNET_HashCode *namespace;
 
   /**
-   * Identity of a peer hosting the content, only set if
-   * 'has_target' is GNUNET_YES.
+   * Identity of a peer hosting the content, otherwise NULl.
    * Allocated after struct only if needed. Do not free!
    */
   const struct GNUNET_PeerIdentity *target;
@@ -360,6 +359,15 @@
 
 
 /**
+ * Consider downloading via stream (if possible)
+ *
+ * @param pr the pending request to process
+ */
+void
+GSF_stream_lookup_ (struct GSF_PendingRequest *pr);
+
+
+/**
  * Function to be called after we're done processing
  * replies from the local lookup.
  *

Modified: gnunet/src/fs/gnunet-service-fs_put.c
===================================================================
--- gnunet/src/fs/gnunet-service-fs_put.c       2012-11-24 23:25:49 UTC (rev 
25122)
+++ gnunet/src/fs/gnunet-service-fs_put.c       2012-11-24 23:45:27 UTC (rev 
25123)
@@ -180,8 +180,11 @@
  *        maybe 0 if no unique identifier is available
  */
 static void
-process_dht_put_content (void *cls, const struct GNUNET_HashCode * key, size_t 
size,
-                         const void *data, enum GNUNET_BLOCK_Type type,
+process_dht_put_content (void *cls, 
+                        const struct GNUNET_HashCode * key, 
+                        size_t size,
+                         const void *data, 
+                        enum GNUNET_BLOCK_Type type,
                          uint32_t priority, uint32_t anonymity,
                          struct GNUNET_TIME_Absolute expiration, uint64_t uid)
 {