proper datacache expiration by proximity first

4 files changed, 93 insertions, 91 deletions

diff --git a/src/datacache/datacache.c b/src/datacache/datacache.c
index 18a2ed228..7c0a975da 100644
--- a/src/datacache/datacache.c
+++ b/src/datacache/datacache.c
@@ -260,7 +260,7 @@ GNUNET_DATACACHE_destroy (struct GNUNET_DATACACHE_Handle *h)
  *
  * @param h handle to the datacache
  * @param key key to store data under
- * @param am_closest are we the closest peer?
+ * @param xor_distance distance of @a key to our PID
  * @param data_size number of bytes in @a data
  * @param data data to store
  * @param type type of the value
@@ -272,7 +272,7 @@ GNUNET_DATACACHE_destroy (struct GNUNET_DATACACHE_Handle *h)
 int
 GNUNET_DATACACHE_put (struct GNUNET_DATACACHE_Handle *h,
                       const struct GNUNET_HashCode *key,
-                      int am_closest,
+                      uint32_t xor_distance,
                       size_t data_size,
                       const char *data,
                       enum GNUNET_BLOCK_Type type,
@@ -284,7 +284,7 @@ GNUNET_DATACACHE_put (struct GNUNET_DATACACHE_Handle *h,
 
   used = h->api->put (h->api->cls,
                       key,
-                      am_closest,
+                      xor_distance,
                       data_size,
                       data,
                       type,
diff --git a/src/datacache/plugin_datacache_heap.c b/src/datacache/plugin_datacache_heap.c
index c32edf8e2..ad5e7834d 100644
--- a/src/datacache/plugin_datacache_heap.c
+++ b/src/datacache/plugin_datacache_heap.c
@@ -31,7 +31,7 @@
 
 #define LOG_STRERROR_FILE(kind,op,fn) GNUNET_log_from_strerror_file (kind, "datacache-heap", op, fn)
 
-
+#define NUM_HEAPS 24
 
 /**
  * Context for all functions in this plugin.
@@ -49,14 +49,9 @@ struct Plugin
   struct GNUNET_CONTAINER_MultiHashMap *map;
 
   /**
-   * Heap for expirations.
+   * Heaps sorted by distance.
    */
-  struct GNUNET_CONTAINER_Heap *heap;
-
-  /**
-   * Heap from the plugin for "closest" values.
-   */
-  struct GNUNET_CONTAINER_Heap *cheap;
+  struct GNUNET_CONTAINER_Heap *heaps[NUM_HEAPS];
 
 };
 
@@ -97,9 +92,9 @@ struct Value
   unsigned int path_info_len;
 
   /**
-   * Am I the closest peer? Determines which heap we are in!
+   * How close is the hash to us? Determines which heap we are in!
    */
-  int am_closest;
+  uint32_t distance;
 
   /**
    * Type of the block.
@@ -128,16 +123,6 @@ struct PutContext
   const char *data;
 
   /**
-   * Heap from the plugin for "closest" values.
-   */
-  struct GNUNET_CONTAINER_Heap *cheap;
-
-  /**
-   * Heap from the plugin.
-   */
-  struct GNUNET_CONTAINER_Heap *heap;
-
-  /**
    * Path information.
    */
   const struct GNUNET_PeerIdentity *path_info;
@@ -195,8 +180,8 @@ put_cb (void *cls,
                        val->path_info_len,
                        put_ctx->path_info_len);
     GNUNET_memcpy (val->path_info,
-            put_ctx->path_info,
-            put_ctx->path_info_len * sizeof (struct GNUNET_PeerIdentity));
+                   put_ctx->path_info,
+                   put_ctx->path_info_len * sizeof (struct GNUNET_PeerIdentity));
     GNUNET_CONTAINER_heap_update_cost (val->hn,
                                        val->discard_time.abs_value_us);
     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
@@ -216,7 +201,7 @@ put_cb (void *cls,
  *
  * @param cls closure (our `struct Plugin`)
  * @param key key to store data under
- * @param am_closest are we the closest peer?
+ * @param xor_distance how close is @a key to our PID?
  * @param size number of bytes in @a data
  * @param data data to store
  * @param type type of the value
@@ -228,7 +213,7 @@ put_cb (void *cls,
 static ssize_t
 heap_plugin_put (void *cls,
                  const struct GNUNET_HashCode *key,
-                 int am_closest,
+                 uint32_t xor_distance,
                  size_t size,
                  const char *data,
                  enum GNUNET_BLOCK_Type type,
@@ -241,8 +226,6 @@ heap_plugin_put (void *cls,
   struct PutContext put_ctx;
 
   put_ctx.found = GNUNET_NO;
-  put_ctx.heap = plugin->heap;
-  put_ctx.cheap = plugin->cheap;
   put_ctx.data = data;
   put_ctx.size = size;
   put_ctx.path_info = path_info;
@@ -256,12 +239,17 @@ heap_plugin_put (void *cls,
   if (GNUNET_YES == put_ctx.found)
     return 0;
   val = GNUNET_malloc (sizeof (struct Value) + size);
-  GNUNET_memcpy (&val[1], data, size);
+  GNUNET_memcpy (&val[1],
+                 data,
+                 size);
   val->key = *key;
   val->type = type;
   val->discard_time = discard_time;
   val->size = size;
-  val->am_closest = am_closest;
+  if (xor_distance >= NUM_HEAPS)
+    val->distance = NUM_HEAPS - 1;
+  else
+    val->distance = xor_distance;
   GNUNET_array_grow (val->path_info,
                      val->path_info_len,
                      path_info_len);
@@ -272,9 +260,7 @@ heap_plugin_put (void *cls,
                                             &val->key,
                                             val,
                                             GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
-  val->hn = GNUNET_CONTAINER_heap_insert (am_closest
-                                          ? plugin->cheap
-                                          : plugin->heap,
+  val->hn = GNUNET_CONTAINER_heap_insert (plugin->heaps[val->distance],
                                           val,
                                           val->discard_time.abs_value_us);
   return size + OVERHEAD;
@@ -392,9 +378,12 @@ heap_plugin_del (void *cls)
   struct Plugin *plugin = cls;
   struct Value *val;
 
-  val = GNUNET_CONTAINER_heap_remove_root (plugin->heap);
-  if (NULL == val)
-    val = GNUNET_CONTAINER_heap_remove_root (plugin->cheap);
+  for (unsigned int i=0;i<NUM_HEAPS;i++)
+  {
+    val = GNUNET_CONTAINER_heap_remove_root (plugin->heaps[i]);
+    if (NULL != val)
+      break;
+  }
   if (NULL == val)
     return GNUNET_SYSERR;
   GNUNET_assert (GNUNET_YES ==
@@ -547,8 +536,8 @@ libgnunet_plugin_datacache_heap_init (void *cls)
   plugin = GNUNET_new (struct Plugin);
   plugin->map = GNUNET_CONTAINER_multihashmap_create (1024,  /* FIXME: base on quota! */
                                                       GNUNET_YES);
-  plugin->heap = GNUNET_CONTAINER_heap_create (GNUNET_CONTAINER_HEAP_ORDER_MIN);
-  plugin->cheap = GNUNET_CONTAINER_heap_create (GNUNET_CONTAINER_HEAP_ORDER_MIN);
+  for (unsigned int i=0;i<NUM_HEAPS;i++)
+    plugin->heaps[i] = GNUNET_CONTAINER_heap_create (GNUNET_CONTAINER_HEAP_ORDER_MIN);
   plugin->env = env;
   api = GNUNET_new (struct GNUNET_DATACACHE_PluginFunctions);
   api->cls = plugin;
@@ -576,26 +565,19 @@ libgnunet_plugin_datacache_heap_done (void *cls)
   struct Plugin *plugin = api->cls;
   struct Value *val;
 
-  while (NULL != (val = GNUNET_CONTAINER_heap_remove_root (plugin->heap)))
+  for (unsigned int i=0;i<NUM_HEAPS;i++)
   {
-    GNUNET_assert (GNUNET_YES ==
-                   GNUNET_CONTAINER_multihashmap_remove (plugin->map,
-                                                         &val->key,
-                                                         val));
-    GNUNET_free_non_null (val->path_info);
-    GNUNET_free (val);
-  }
-  while (NULL != (val = GNUNET_CONTAINER_heap_remove_root (plugin->cheap)))
-  {
-    GNUNET_assert (GNUNET_YES ==
-                   GNUNET_CONTAINER_multihashmap_remove (plugin->map,
-                                                         &val->key,
-                                                         val));
-    GNUNET_free_non_null (val->path_info);
-    GNUNET_free (val);
+    while (NULL != (val = GNUNET_CONTAINER_heap_remove_root (plugin->heaps[i])))
+    {
+      GNUNET_assert (GNUNET_YES ==
+                     GNUNET_CONTAINER_multihashmap_remove (plugin->map,
+                                                           &val->key,
+                                                           val));
+      GNUNET_free_non_null (val->path_info);
+      GNUNET_free (val);
+    }
+    GNUNET_CONTAINER_heap_destroy (plugin->heaps[i]);
   }
-  GNUNET_CONTAINER_heap_destroy (plugin->heap);
-  GNUNET_CONTAINER_heap_destroy (plugin->cheap);
   GNUNET_CONTAINER_multihashmap_destroy (plugin->map);
   GNUNET_free (plugin);
   GNUNET_free (api);
diff --git a/src/datacache/plugin_datacache_sqlite.c b/src/datacache/plugin_datacache_sqlite.c
index 455dcab0b..c4adb34fd 100644
--- a/src/datacache/plugin_datacache_sqlite.c
+++ b/src/datacache/plugin_datacache_sqlite.c
@@ -83,6 +83,11 @@ struct Plugin
   /**
    * Prepared statement for #sqlite_plugin_del.
    */
+  sqlite3_stmt *del_expired_stmt;
+
+  /**
+   * Prepared statement for #sqlite_plugin_del.
+   */
   sqlite3_stmt *del_stmt;
 
   /**
@@ -150,7 +155,7 @@ sq_prepare (sqlite3 *dbh,
  *
  * @param cls closure (our `struct Plugin`)
  * @param key key to store @a data under
- * @param am_closest are we the closest peer?
+ * @param xor_distance how close is @a key to our PID?
  * @param size number of bytes in @a data
  * @param data data to store
  * @param type type of the value
@@ -162,7 +167,7 @@ sq_prepare (sqlite3 *dbh,
 static ssize_t
 sqlite_plugin_put (void *cls,
                    const struct GNUNET_HashCode *key,
-                   int am_closest,
+                   uint32_t xor_distance,
                    size_t size,
                    const char *data,
                    enum GNUNET_BLOCK_Type type,
@@ -172,12 +177,11 @@ sqlite_plugin_put (void *cls,
 {
   struct Plugin *plugin = cls;
   uint32_t type32 = type;
-  uint32_t prox = am_closest;
   struct GNUNET_SQ_QueryParam params[] = {
     GNUNET_SQ_query_param_uint32 (&type32),
     GNUNET_SQ_query_param_absolute_time (&discard_time),
     GNUNET_SQ_query_param_auto_from_type (key),
-    GNUNET_SQ_query_param_uint32 (&prox),
+    GNUNET_SQ_query_param_uint32 (&xor_distance),
     GNUNET_SQ_query_param_fixed_size (data, size),
     GNUNET_SQ_query_param_fixed_size (path_info,
                                       path_info_len * sizeof (struct GNUNET_PeerIdentity)),
@@ -390,8 +394,8 @@ sqlite_plugin_del (void *cls)
   uint64_t rowid;
   void *data;
   size_t dsize;
-  uint32_t prox;
   struct GNUNET_HashCode hc;
+  struct GNUNET_TIME_Absolute now;
   struct GNUNET_SQ_ResultSpec rs[] = {
     GNUNET_SQ_result_spec_uint64 (&rowid),
     GNUNET_SQ_result_spec_auto_from_type (&hc),
@@ -403,54 +407,61 @@ sqlite_plugin_del (void *cls)
     GNUNET_SQ_query_param_uint64 (&rowid),
     GNUNET_SQ_query_param_end
   };
-  struct GNUNET_SQ_QueryParam prox_params[] = {
-    GNUNET_SQ_query_param_uint32 (&prox),
+  struct GNUNET_SQ_QueryParam time_params[] = {
+    GNUNET_SQ_query_param_absolute_time (&now),
     GNUNET_SQ_query_param_end
   };
 
   LOG (GNUNET_ERROR_TYPE_DEBUG,
        "Processing DEL\n");
-  prox = GNUNET_NO;
- again:
+  now = GNUNET_TIME_absolute_get ();
   if (GNUNET_OK !=
-      GNUNET_SQ_bind (plugin->del_select_stmt,
-                      prox_params))
+      GNUNET_SQ_bind (plugin->del_expired_stmt,
+                      time_params))
   {
     LOG_SQLITE (plugin->dbh,
                 GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
                 "sqlite3_bind");
     GNUNET_SQ_reset (plugin->dbh,
-                     plugin->del_stmt);
+                     plugin->del_expired_stmt);
     return GNUNET_SYSERR;
   }
   if (SQLITE_ROW !=
-      sqlite3_step (plugin->del_select_stmt))
+      sqlite3_step (plugin->del_expired_stmt))
   {
     LOG_SQLITE (plugin->dbh,
                 GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
                 "sqlite3_step");
     GNUNET_SQ_reset (plugin->dbh,
-                     plugin->del_select_stmt);
-    if (GNUNET_NO == prox)
-    {
-      prox = GNUNET_YES;
-      goto again;
-    }
+                     plugin->del_expired_stmt);
     return GNUNET_SYSERR;
   }
   if (GNUNET_OK !=
-      GNUNET_SQ_extract_result (plugin->del_select_stmt,
+      GNUNET_SQ_extract_result (plugin->del_expired_stmt,
                                 rs))
   {
     GNUNET_SQ_reset (plugin->dbh,
-                     plugin->del_select_stmt);
-    if (GNUNET_NO == prox)
+                     plugin->del_expired_stmt);
+
+    if (SQLITE_ROW !=
+        sqlite3_step (plugin->del_select_stmt))
     {
-      prox = GNUNET_YES;
-      goto again;
+      LOG_SQLITE (plugin->dbh,
+                  GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
+                  "sqlite3_step");
+      GNUNET_SQ_reset (plugin->dbh,
+                       plugin->del_select_stmt);
+      return GNUNET_SYSERR;
+    }
+    if (GNUNET_OK !=
+        GNUNET_SQ_extract_result (plugin->del_select_stmt,
+                                  rs))
+    {
+      GNUNET_SQ_reset (plugin->dbh,
+                       plugin->del_select_stmt);
+      GNUNET_break (0);
+      return GNUNET_SYSERR;
     }
-    GNUNET_break (0);
-    return GNUNET_SYSERR;
   }
   GNUNET_SQ_cleanup_result (rs);
   GNUNET_SQ_reset (plugin->dbh,
@@ -741,14 +752,15 @@ libgnunet_plugin_datacache_sqlite_init (void *cls)
     SQLITE3_EXEC (dbh, "PRAGMA sqlite_temp_store=3");
 
   SQLITE3_EXEC (dbh,
-                "CREATE TABLE ds091 (" "  type INTEGER NOT NULL DEFAULT 0,"
+                "CREATE TABLE ds091 ("
+                "  type INTEGER NOT NULL DEFAULT 0,"
                 "  expire INTEGER NOT NULL,"
                 "  key BLOB NOT NULL DEFAULT '',"
                 "  prox INTEGER NOT NULL,"
                 "  value BLOB NOT NULL,"
                 "  path BLOB DEFAULT '')");
-  SQLITE3_EXEC (dbh, "CREATE INDEX idx_hashidx ON ds090 (key,type,expire)");
-  SQLITE3_EXEC (dbh, "CREATE INDEX idx_expire ON ds090 (prox,expire)");
+  SQLITE3_EXEC (dbh, "CREATE INDEX idx_hashidx ON ds091 (key,type,expire)");
+  SQLITE3_EXEC (dbh, "CREATE INDEX idx_expire ON ds091 (prox,expire)");
   plugin = GNUNET_new (struct Plugin);
   plugin->env = env;
   plugin->dbh = dbh;
@@ -766,12 +778,19 @@ libgnunet_plugin_datacache_sqlite_init (void *cls)
                     &plugin->get_count_stmt)) ||
        (SQLITE_OK !=
         sq_prepare (plugin->dbh,
-                    "SELECT value,expire,path FROM ds091 "
-                    "WHERE key=? AND type=? AND expire >= ? LIMIT 1 OFFSET ?",
+                    "SELECT value,expire,path FROM ds091"
+                    " WHERE key=? AND type=? AND expire >= ? LIMIT 1 OFFSET ?",
                     &plugin->get_stmt)) ||
        (SQLITE_OK !=
         sq_prepare (plugin->dbh,
-                    "SELECT _ROWID_,key,value FROM ds091 WHERE prox=? ORDER BY expire ASC LIMIT 1",
+                    "SELECT _ROWID_,key,value FROM ds091"
+                    " WHERE expire < ?"
+                    " ORDER BY expire ASC LIMIT 1",
+                    &plugin->del_expired_stmt)) ||
+       (SQLITE_OK !=
+        sq_prepare (plugin->dbh,
+                    "SELECT _ROWID_,key,value FROM ds091"
+                    " ORDER BY prox ASC, expire ASC LIMIT 1",
                     &plugin->del_select_stmt)) ||
        (SQLITE_OK !=
         sq_prepare (plugin->dbh,
@@ -840,6 +859,7 @@ libgnunet_plugin_datacache_sqlite_done (void *cls)
   sqlite3_finalize (plugin->get_count_stmt);
   sqlite3_finalize (plugin->get_stmt);
   sqlite3_finalize (plugin->del_select_stmt);
+  sqlite3_finalize (plugin->del_expired_stmt);
   sqlite3_finalize (plugin->del_stmt);
   sqlite3_finalize (plugin->get_random_stmt);
   sqlite3_finalize (plugin->get_closest_stmt);
diff --git a/src/datacache/plugin_datacache_template.c b/src/datacache/plugin_datacache_template.c
index 28bcbcd26..1064d3125 100644
--- a/src/datacache/plugin_datacache_template.c
+++ b/src/datacache/plugin_datacache_template.c
@@ -45,7 +45,7 @@ struct Plugin
  *
  * @param cls closure (our `struct Plugin`)
  * @param key key to store @a data under
- * @param am_closest are we the closest peer?
+ * @param xor_distance distance of @a key to our PID
  * @param size number of bytes in @a data
  * @param data data to store
  * @param type type of the value
@@ -57,7 +57,7 @@ struct Plugin
 static ssize_t
 template_plugin_put (void *cls,
                      const struct GNUNET_HashCode *key,
-                     int am_closest,
+                     uint32_t xor_distance,
                      size_t size,
                      const char *data,
                      enum GNUNET_BLOCK_Type type,