[GNUnet-SVN] r6395 - in GNUnet: . src/applications/fs/gap src/applications/sqstore_mysql

[gnunet]

Author: grothoff
Date: 2008-02-20 21:56:19 -0700 (Wed, 20 Feb 2008)
New Revision: 6395

Modified:
   GNUnet/src/applications/fs/gap/gap.c
   GNUnet/src/applications/fs/gap/plan.c
   GNUnet/src/applications/fs/gap/shared.h
   GNUnet/src/applications/sqstore_mysql/mysql.c
   GNUnet/todo
Log:
push out more replies

Modified: GNUnet/src/applications/fs/gap/gap.c
===================================================================
--- GNUnet/src/applications/fs/gap/gap.c        2008-02-21 03:30:35 UTC (rev 
6394)
+++ GNUnet/src/applications/fs/gap/gap.c        2008-02-21 04:56:19 UTC (rev 
6395)
@@ -43,6 +43,17 @@
 #define MAX_ENTRIES_PER_SLOT 2
 
 /**
+ * If, after finding local results, we abort a GET
+ * iteration, we increment "have_more" by this value.
+ */
+#define HAVE_MORE_INCREMENT 5
+
+/**
+ * How often do we check have_more?
+ */
+#define HAVE_MORE_FREQUENCY (100 * GNUNET_CRON_MILLISECONDS)
+
+/**
  * The GAP routing table.
  */
 static struct RequestList **table;
@@ -149,7 +160,11 @@
   want_more = GNUNET_OK;
   cls->iteration_count++;
   if (cls->iteration_count > 10 * (1 + req->value))
-    want_more = GNUNET_SYSERR;
+    {
+      if (cls->result_count > 0)
+        req->have_more += HAVE_MORE_INCREMENT;
+      want_more = GNUNET_SYSERR;
+    }
   enc = NULL;
   if (ntohl (value->type) == GNUNET_ECRS_BLOCKTYPE_ONDEMAND)
     {
@@ -184,7 +199,10 @@
   memcpy (&msg[1], &value[1], size - sizeof (P2P_gap_reply_MESSAGE));
   cls->result_count++;
   if (cls->result_count > 2 * (1 + req->value))
-    want_more = GNUNET_SYSERR;
+    {
+      req->have_more += HAVE_MORE_INCREMENT;
+      want_more = GNUNET_SYSERR;
+    }
   GNUNET_cron_add_job (cron,
                        send_delayed,
                        GNUNET_random_u32 (GNUNET_RANDOM_QUALITY_WEAK,
@@ -365,11 +383,11 @@
   cls.request = rl;
   cls.iteration_count = 0;
   cls.result_count = 0;
-  ret = datastore->get (&queries[0], type, datastore_value_processor, &cls);
+  ret = datastore->get (&queries[0], type, &datastore_value_processor, &cls);
   if ((type == GNUNET_ECRS_BLOCKTYPE_DATA) && (ret != 1))
     ret = datastore->get (&queries[0],
                           GNUNET_ECRS_BLOCKTYPE_ONDEMAND,
-                          datastore_value_processor, &cls);
+                          &datastore_value_processor, &cls);
 
   /* if not found or not unique, forward */
   if (((ret != 1) || (type != GNUNET_ECRS_BLOCKTYPE_DATA)) &&
@@ -556,6 +574,46 @@
   GNUNET_mutex_unlock (GNUNET_FS_lock);
 }
 
+/**
+ * Cron-job to find and transmit more results (beyond
+ * the initial batch) over time -- assuming the entry
+ * is still valid and we have more data.
+ */
+static void
+have_more_processor (void *unused)
+{
+  static unsigned int pos;
+  struct RequestList *req;
+  GNUNET_CronTime now;
+  struct DVPClosure cls;
+
+  GNUNET_mutex_lock (GNUNET_FS_lock);
+  now = GNUNET_get_time ();
+  if (pos >= table_size)
+    pos = 0;
+  req = table[pos];
+  while (req != NULL)
+    {
+      if ((GNUNET_cpu_get_load (coreAPI->ectx,
+                                coreAPI->cfg) > 50) ||
+          (GNUNET_disk_get_load (coreAPI->ectx, coreAPI->cfg) > 25))
+        break;
+      if (req->have_more > 0)
+        {
+          req->have_more--;
+          cls.request = req;
+          cls.iteration_count = 0;
+          cls.result_count = 0;
+          datastore->get (&req->queries[0], req->type,
+                          &datastore_value_processor, &cls);
+        }
+      req = req->next;
+    }
+  if (req == NULL)
+    pos++;
+  GNUNET_mutex_unlock (GNUNET_FS_lock);
+}
+
 int
 GNUNET_FS_GAP_init (GNUNET_CoreAPIForPlugins * capi)
 {
@@ -582,6 +640,10 @@
                     coreAPI->
                     register_notify_peer_disconnect
                     (&cleanup_on_peer_disconnect, NULL));
+  GNUNET_cron_add_job (capi->cron,
+                       &have_more_processor,
+                       HAVE_MORE_FREQUENCY, HAVE_MORE_FREQUENCY, NULL);
+
   stats = capi->request_service ("stats");
   if (stats != NULL)
     {
@@ -608,6 +670,9 @@
   unsigned int i;
   struct RequestList *rl;
 
+  GNUNET_cron_del_job (coreAPI->cron,
+                       &have_more_processor, HAVE_MORE_FREQUENCY, NULL);
+
   for (i = 0; i < table_size; i++)
     {
       while (NULL != (rl = table[i]))

Modified: GNUnet/src/applications/fs/gap/plan.c
===================================================================
--- GNUnet/src/applications/fs/gap/plan.c       2008-02-21 03:30:35 UTC (rev 
6394)
+++ GNUnet/src/applications/fs/gap/plan.c       2008-02-21 04:56:19 UTC (rev 
6395)
@@ -309,8 +309,8 @@
  * @param prio priority to use for the request
  */
 static void
-add_request (PID_INDEX target,
-             struct RequestList *request, int ttl, unsigned int prio)
+queue_request (PID_INDEX target,
+               struct RequestList *request, int ttl, unsigned int prio)
 {
   struct QueryPlanList *qpl;
   struct QueryPlanEntry *entry;
@@ -611,7 +611,7 @@
                     }
                   request->remaining_value -= rank->prio;
                 }
-              add_request (rank->peer, request, rank->ttl, rank->prio);
+              queue_request (rank->peer, request, rank->ttl, rank->prio);
               total_score -= rank->score;
               rank->score = 0;  /* mark as used */
               break;
@@ -677,7 +677,7 @@
     GNUNET_bloomfilter_get_raw_data (req->bloomfilter,
                                      (char *) &msg->queries[req->key_count],
                                      req->bloomfilter_size);
-  now = GNUNET_get_time();
+  now = GNUNET_get_time ();
   if (now + ttl > req->last_request_time + req->last_ttl_used)
     {
       req->last_request_time = now;

Modified: GNUnet/src/applications/fs/gap/shared.h
===================================================================
--- GNUnet/src/applications/fs/gap/shared.h     2008-02-21 03:30:35 UTC (rev 
6394)
+++ GNUnet/src/applications/fs/gap/shared.h     2008-02-21 04:56:19 UTC (rev 
6395)
@@ -180,6 +180,16 @@
   unsigned int remaining_value;
 
   /**
+   * What is the chance that we have more results
+   * locally for this request?  Set to a positive
+   * number if we think we have more results, 
+   * decremented by one each time we fail to find
+   * more results; set to zero if we are sure 
+   * that we have no more results.
+   */
+  unsigned int have_more;
+
+  /**
    * Routing policy for the request (foward, indirect).
    */
   enum GNUNET_FS_RoutingPolicy policy;

Modified: GNUnet/src/applications/sqstore_mysql/mysql.c
===================================================================
--- GNUnet/src/applications/sqstore_mysql/mysql.c       2008-02-21 03:30:35 UTC 
(rev 6394)
+++ GNUnet/src/applications/sqstore_mysql/mysql.c       2008-02-21 04:56:19 UTC 
(rev 6395)
@@ -1485,7 +1485,7 @@
       if (count + off == total)
         last_vkey = 0;          /* back to start */
       if (count == total)
-       break;
+        break;
     }
   mysql_thread_end ();
   return count;

Modified: GNUnet/todo
===================================================================
--- GNUnet/todo 2008-02-21 03:30:35 UTC (rev 6394)
+++ GNUnet/todo 2008-02-21 04:56:19 UTC (rev 6395)
@@ -35,7 +35,9 @@
   to a valid entry, and table[index]->next is NULL at the time of the
   crash.
 
- 
+
+
+
 - test RPC code (write a small demo)
 
 Hostlist for network testing: