implementing 'get_closest' API for sqlite and postgres datacache plugins

4 files changed, 309 insertions, 4 deletions

diff --git a/src/datacache/datacache.c b/src/datacache/datacache.c
index 426841bc8..6c8d2b8cc 100644
--- a/src/datacache/datacache.c
+++ b/src/datacache/datacache.c
@@ -375,4 +375,39 @@ GNUNET_DATACACHE_get_random (struct GNUNET_DATACACHE_Handle *h,
 }
 
 
+/**
+ * Iterate over the results that are "close" to a particular key in
+ * the datacache.  "close" is defined as numerically larger than @a
+ * key (when interpreted as a circular address space), with small
+ * distance.
+ *
+ * @param h handle to the datacache
+ * @param key area of the keyspace to look into
+ * @param num_results number of results that should be returned to @a iter
+ * @param iter maybe NULL (to just count)
+ * @param iter_cls closure for @a iter
+ * @return the number of results found
+ */
+unsigned int
+GNUNET_DATACACHE_get_closest (struct GNUNET_DATACACHE_Handle *h,
+                              const struct GNUNET_HashCode *key,
+                              unsigned int num_results,
+                              GNUNET_DATACACHE_Iterator iter,
+                              void *iter_cls)
+{
+  GNUNET_STATISTICS_update (h->stats,
+                            gettext_noop ("# proximity search requests received"),
+                            1,
+                            GNUNET_NO);
+  LOG (GNUNET_ERROR_TYPE_DEBUG,
+       "Processing proximity search at `%s'\n",
+       GNUNET_h2s (key));
+  return h->api->get_closest (h->api->cls,
+                              key,
+                              num_results,
+                              iter,
+                              iter_cls);
+}
+
+
 /* end of datacache.c */
diff --git a/src/datacache/plugin_datacache_heap.c b/src/datacache/plugin_datacache_heap.c
index 32e762e23..5492a8ec5 100644
--- a/src/datacache/plugin_datacache_heap.c
+++ b/src/datacache/plugin_datacache_heap.c
@@ -413,6 +413,31 @@ heap_plugin_get_random (void *cls,
 
 
 /**
+ * Iterate over the results that are "close" to a particular key in
+ * the datacache.  "close" is defined as numerically larger than @a
+ * key (when interpreted as a circular address space), with small
+ * distance.
+ *
+ * @param cls closure (internal context for the plugin)
+ * @param key area of the keyspace to look into
+ * @param num_results number of results that should be returned to @a iter
+ * @param iter maybe NULL (to just count)
+ * @param iter_cls closure for @a iter
+ * @return the number of results found
+ */
+static unsigned int
+heap_plugin_get_closest (void *cls,
+                         const struct GNUNET_HashCode *key,
+                         unsigned int num_results,
+                         GNUNET_DATACACHE_Iterator iter,
+                         void *iter_cls)
+{
+  GNUNET_break (0); // not implemented!
+  return 0;
+}
+
+
+/**
  * Entry point for the plugin.
  *
  * @param cls closure (the `struct GNUNET_DATACACHE_PluginEnvironmnet`)
@@ -436,6 +461,7 @@ libgnunet_plugin_datacache_heap_init (void *cls)
   api->put = &heap_plugin_put;
   api->del = &heap_plugin_del;
   api->get_random = &heap_plugin_get_random;
+  api->get_closest = &heap_plugin_get_closest;
   LOG (GNUNET_ERROR_TYPE_INFO,
        _("Heap datacache running\n"));
   return api;
diff --git a/src/datacache/plugin_datacache_postgres.c b/src/datacache/plugin_datacache_postgres.c
index 13f944d7c..1156d214e 100644
--- a/src/datacache/plugin_datacache_postgres.c
+++ b/src/datacache/plugin_datacache_postgres.c
@@ -163,6 +163,11 @@ init_connection (struct Plugin *plugin)
                                 "ORDER BY key ASC LIMIT 1 OFFSET $1", 1)) ||
       (GNUNET_OK !=
        GNUNET_POSTGRES_prepare (plugin->dbh,
+                                "get_closest",
+                                "SELECT discard_time,type,value,path,key FROM gn090dc "
+                                "WHERE key>=$1 ORDER BY key ASC LIMIT $2", 1)) ||
+      (GNUNET_OK !=
+       GNUNET_POSTGRES_prepare (plugin->dbh,
                                 "delrow",
                                 "DELETE FROM gn090dc WHERE oid=$1", 1)) ||
       (GNUNET_OK !=
@@ -259,11 +264,11 @@ postgres_plugin_get (void *cls,
 
   const char *paramValues[] = {
     (const char *) key,
-    (const char *) &btype,
+    (const char *) &btype
   };
   int paramLengths[] = {
     sizeof (struct GNUNET_HashCode),
-    sizeof (btype),
+    sizeof (btype)
   };
   const int paramFormats[] = { 1, 1 };
   struct GNUNET_TIME_Absolute expiration_time;
@@ -433,10 +438,10 @@ postgres_plugin_get_random (void *cls,
   unsigned int type;
   PGresult *res;
   const char *paramValues[] = {
-    (const char *) &off_be,
+    (const char *) &off_be
   };
   int paramLengths[] = {
-    sizeof (off_be),
+    sizeof (off_be)
   };
   const int paramFormats[] = { 1 };
 
@@ -507,6 +512,129 @@ postgres_plugin_get_random (void *cls,
 
 
 /**
+ * Iterate over the results that are "close" to a particular key in
+ * the datacache.  "close" is defined as numerically larger than @a
+ * key (when interpreted as a circular address space), with small
+ * distance.
+ *
+ * @param cls closure (internal context for the plugin)
+ * @param key area of the keyspace to look into
+ * @param num_results number of results that should be returned to @a iter
+ * @param iter maybe NULL (to just count)
+ * @param iter_cls closure for @a iter
+ * @return the number of results found
+ */
+static unsigned int
+postgres_plugin_get_closest (void *cls,
+                             const struct GNUNET_HashCode *key,
+                             unsigned int num_results,
+                             GNUNET_DATACACHE_Iterator iter,
+                             void *iter_cls)
+{
+  struct Plugin *plugin = cls;
+  uint32_t nbo_limit = htonl (num_results);
+  const char *paramValues[] = {
+    (const char *) key,
+    (const char *) &nbo_limit,
+  };
+  int paramLengths[] = {
+    sizeof (struct GNUNET_HashCode),
+    sizeof (nbo_limit)
+
+  };
+  const int paramFormats[] = { 1, 1 };
+  struct GNUNET_TIME_Absolute expiration_time;
+  uint32_t size;
+  unsigned int type;
+  unsigned int cnt;
+  unsigned int i;
+  unsigned int path_len;
+  const struct GNUNET_PeerIdentity *path;
+  PGresult *res;
+
+  res =
+      PQexecPrepared (plugin->dbh,
+                      "get_closest",
+                      2,
+                      paramValues,
+                      paramLengths,
+                      paramFormats,
+                      1);
+  if (GNUNET_OK !=
+      GNUNET_POSTGRES_check_result (plugin->dbh,
+                                    res,
+                                    PGRES_TUPLES_OK,
+                                    "PQexecPrepared",
+                                    "get_closest"))
+  {
+    LOG (GNUNET_ERROR_TYPE_DEBUG,
+         "Ending iteration (postgres error)\n");
+    return 0;
+  }
+
+  if (0 == (cnt = PQntuples (res)))
+  {
+    /* no result */
+    LOG (GNUNET_ERROR_TYPE_DEBUG,
+         "Ending iteration (no more results)\n");
+    PQclear (res);
+    return 0;
+  }
+  if (NULL == iter)
+  {
+    PQclear (res);
+    return cnt;
+  }
+  if ( (5 != PQnfields (res)) ||
+       (sizeof (uint64_t) != PQfsize (res, 0)) ||
+       (sizeof (uint32_t) != PQfsize (res, 1)) ||
+       (sizeof (struct GNUNET_HashCode) != PQfsize (res, 4)) )
+  {
+    GNUNET_break (0);
+    PQclear (res);
+    return 0;
+  }
+  for (i = 0; i < cnt; i++)
+  {
+    expiration_time.abs_value_us =
+        GNUNET_ntohll (*(uint64_t *) PQgetvalue (res, i, 0));
+    type = ntohl (*(uint32_t *) PQgetvalue (res, i, 1));
+    size = PQgetlength (res, i, 2);
+    path_len = PQgetlength (res, i, 3);
+    if (0 != (path_len % sizeof (struct GNUNET_PeerIdentity)))
+    {
+      GNUNET_break (0);
+      path_len = 0;
+    }
+    path_len %= sizeof (struct GNUNET_PeerIdentity);
+    path = (const struct GNUNET_PeerIdentity *) PQgetvalue (res, i, 3);
+    key = (const struct GNUNET_HashCode *) PQgetvalue (res, i, 4);
+    LOG (GNUNET_ERROR_TYPE_DEBUG,
+         "Found result of size %u bytes and type %u in database\n",
+         (unsigned int) size,
+         (unsigned int) type);
+    if (GNUNET_SYSERR ==
+        iter (iter_cls,
+              key,
+              size,
+              PQgetvalue (res, i, 2),
+              (enum GNUNET_BLOCK_Type) type,
+              expiration_time,
+              path_len,
+              path))
+    {
+      LOG (GNUNET_ERROR_TYPE_DEBUG,
+           "Ending iteration (client error)\n");
+      PQclear (res);
+      return cnt;
+    }
+  }
+  PQclear (res);
+  return cnt;
+}
+
+
+/**
  * Entry point for the plugin.
  *
  * @param cls closure (the `struct GNUNET_DATACACHE_PluginEnvironmnet`)
@@ -534,6 +662,7 @@ libgnunet_plugin_datacache_postgres_init (void *cls)
   api->put = &postgres_plugin_put;
   api->del = &postgres_plugin_del;
   api->get_random = &postgres_plugin_get_random;
+  api->get_closest = &postgres_plugin_get_closest;
   LOG (GNUNET_ERROR_TYPE_INFO,
        "Postgres datacache running\n");
   return api;
diff --git a/src/datacache/plugin_datacache_sqlite.c b/src/datacache/plugin_datacache_sqlite.c
index d28233772..8dd38a3ac 100644
--- a/src/datacache/plugin_datacache_sqlite.c
+++ b/src/datacache/plugin_datacache_sqlite.c
@@ -522,6 +522,120 @@ sqlite_plugin_get_random (void *cls,
 
 
 /**
+ * Iterate over the results that are "close" to a particular key in
+ * the datacache.  "close" is defined as numerically larger than @a
+ * key (when interpreted as a circular address space), with small
+ * distance.
+ *
+ * @param cls closure (internal context for the plugin)
+ * @param key area of the keyspace to look into
+ * @param num_results number of results that should be returned to @a iter
+ * @param iter maybe NULL (to just count)
+ * @param iter_cls closure for @a iter
+ * @return the number of results found
+ */
+static unsigned int
+sqlite_plugin_get_closest (void *cls,
+                           const struct GNUNET_HashCode *key,
+                           unsigned int num_results,
+                           GNUNET_DATACACHE_Iterator iter,
+                           void *iter_cls)
+{
+  struct Plugin *plugin = cls;
+  sqlite3_stmt *stmt;
+  struct GNUNET_TIME_Absolute now;
+  struct GNUNET_TIME_Absolute exp;
+  unsigned int size;
+  const char *dat;
+  unsigned int cnt;
+  unsigned int psize;
+  unsigned int type;
+  int64_t ntime;
+  const struct GNUNET_PeerIdentity *path;
+
+  now = GNUNET_TIME_absolute_get ();
+  LOG (GNUNET_ERROR_TYPE_DEBUG,
+       "Processing GET_CLOSEST for key `%4s'\n",
+       GNUNET_h2s (key));
+  if (SQLITE_OK !=
+      sq_prepare (plugin->dbh,
+                  "SELECT value,expire,path,type,key FROM ds090 WHERE key>=? AND expire >= ? ORDER BY KEY ASC LIMIT ?",
+                  &stmt))
+  {
+    LOG_SQLITE (plugin->dbh,
+                GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
+                "sq_prepare");
+    return 0;
+  }
+  ntime = (int64_t) now.abs_value_us;
+  GNUNET_assert (ntime >= 0);
+  if ((SQLITE_OK !=
+       sqlite3_bind_blob (stmt,
+                          1,
+                          key,
+                          sizeof (struct GNUNET_HashCode),
+                          SQLITE_TRANSIENT)) ||
+      (SQLITE_OK != sqlite3_bind_int64 (stmt, 2, now.abs_value_us)) ||
+      (SQLITE_OK != sqlite3_bind_int (stmt, 3, num_results)) )
+  {
+    LOG_SQLITE (plugin->dbh,
+                GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
+                "sqlite3_bind_xxx");
+    sqlite3_finalize (stmt);
+    return 0;
+  }
+  cnt = 0;
+  while (SQLITE_ROW == sqlite3_step (stmt))
+  {
+    if (sizeof (struct GNUNET_HashCode) !=
+        sqlite3_column_bytes (stmt, 4))
+    {
+      GNUNET_break (0);
+      break;
+    }
+    size = sqlite3_column_bytes (stmt, 0);
+    dat = sqlite3_column_blob (stmt, 0);
+    exp.abs_value_us = sqlite3_column_int64 (stmt, 1);
+    psize = sqlite3_column_bytes (stmt, 2);
+    type = sqlite3_column_int (stmt, 3);
+    key = sqlite3_column_blob (stmt, 4);
+    if (0 != psize % sizeof (struct GNUNET_PeerIdentity))
+    {
+      GNUNET_break (0);
+      psize = 0;
+    }
+    psize /= sizeof (struct GNUNET_PeerIdentity);
+    if (0 != psize)
+      path = sqlite3_column_blob (stmt, 2);
+    else
+      path = NULL;
+    ntime = (int64_t) exp.abs_value_us;
+    if (ntime == INT64_MAX)
+      exp = GNUNET_TIME_UNIT_FOREVER_ABS;
+    cnt++;
+    LOG (GNUNET_ERROR_TYPE_DEBUG,
+         "Found %u-byte result at %s when processing GET_CLOSE\n",
+         (unsigned int) size,
+         GNUNET_h2s (key));
+    if (GNUNET_OK != iter (iter_cls,
+                           key,
+                           size,
+                           dat,
+                           type,
+                           exp,
+                           psize,
+                           path))
+    {
+      sqlite3_finalize (stmt);
+      break;
+    }
+  }
+  sqlite3_finalize (stmt);
+  return cnt;
+}
+
+
+/**
  * Entry point for the plugin.
  *
  * @param cls closure (the `struct GNUNET_DATACACHE_PluginEnvironment`)
@@ -595,6 +709,7 @@ libgnunet_plugin_datacache_sqlite_init (void *cls)
   api->put = &sqlite_plugin_put;
   api->del = &sqlite_plugin_del;
   api->get_random = &sqlite_plugin_get_random;
+  api->get_closest = &sqlite_plugin_get_closest;
   LOG (GNUNET_ERROR_TYPE_INFO,
        "Sqlite datacache running\n");
   return api;