[datastore] Return and update replication

This fixes a couple FIXMEs in the datastore code. The replication value
is now returned from the datastore and the update function can increase
the replication.

1 files changed, 84 insertions, 33 deletions

diff --git a/src/datastore/plugin_datastore_mysql.c b/src/datastore/plugin_datastore_mysql.c
index 5ae4485cb..6f2a76499 100644
--- a/src/datastore/plugin_datastore_mysql.c
+++ b/src/datastore/plugin_datastore_mysql.c
@@ -150,22 +150,56 @@ struct Plugin
 #define DELETE_ENTRY_BY_UID "DELETE FROM gn090 WHERE uid=?"
   struct GNUNET_MYSQL_StatementHandle *delete_entry_by_uid;
 
-#define SELECT_ENTRY "SELECT type,prio,anonLevel,expire,hash,value,uid FROM gn090 WHERE uid >= ? AND (rvalue >= ? OR 0 = ?) ORDER BY uid LIMIT 1"
+#define RESULT_COLUMNS "repl, type, prio, anonLevel, expire, hash, value, uid"
+
+#define SELECT_ENTRY "SELECT " RESULT_COLUMNS " FROM gn090 "\
+  "WHERE uid >= ? AND "\
+  "(rvalue >= ? OR 0 = ?) "\
+  "ORDER BY uid LIMIT 1"
   struct GNUNET_MYSQL_StatementHandle *select_entry;
 
-#define SELECT_ENTRY_BY_HASH "SELECT type,prio,anonLevel,expire,hash,value,uid FROM gn090 FORCE INDEX (idx_hash) WHERE hash=? AND uid >= ? AND (rvalue >= ? OR 0 = ?) ORDER BY uid LIMIT 1"
+#define SELECT_ENTRY_BY_HASH "SELECT " RESULT_COLUMNS " FROM gn090 "\
+  "FORCE INDEX (idx_hash) "\
+  "WHERE hash=? AND "\
+  "uid >= ? AND "\
+  "(rvalue >= ? OR 0 = ?) "\
+  "ORDER BY uid LIMIT 1"
   struct GNUNET_MYSQL_StatementHandle *select_entry_by_hash;
 
-#define SELECT_ENTRY_BY_HASH_AND_VHASH "SELECT type,prio,anonLevel,expire,hash,value,uid FROM gn090 FORCE INDEX (idx_hash_vhash) WHERE hash=? AND vhash=? AND uid >= ? AND (rvalue >= ? OR 0 = ?) ORDER BY uid LIMIT 1"
+#define SELECT_ENTRY_BY_HASH_AND_VHASH "SELECT " RESULT_COLUMNS " FROM gn090 "\
+  "FORCE INDEX (idx_hash_vhash) "\
+  "WHERE hash = ? AND "\
+  "vhash = ? AND "\
+  "uid >= ? AND "\
+  "(rvalue >= ? OR 0 = ?) "\
+  "ORDER BY uid LIMIT 1"
   struct GNUNET_MYSQL_StatementHandle *select_entry_by_hash_and_vhash;
 
-#define SELECT_ENTRY_BY_HASH_AND_TYPE "SELECT type,prio,anonLevel,expire,hash,value,uid FROM gn090 FORCE INDEX (idx_hash_type_uid) WHERE hash=? AND type=? AND uid >= ? AND (rvalue >= ? OR 0 = ?) ORDER BY uid LIMIT 1"
+#define SELECT_ENTRY_BY_HASH_AND_TYPE "SELECT " RESULT_COLUMNS " FROM gn090 "\
+  "FORCE INDEX (idx_hash_type_uid) "\
+  "WHERE hash = ? AND "\
+  "type = ? AND "\
+  "uid >= ? AND "\
+  "(rvalue >= ? OR 0 = ?) "\
+  "ORDER BY uid LIMIT 1"
   struct GNUNET_MYSQL_StatementHandle *select_entry_by_hash_and_type;
 
-#define SELECT_ENTRY_BY_HASH_VHASH_AND_TYPE "SELECT type,prio,anonLevel,expire,hash,value,uid FROM gn090 FORCE INDEX (idx_hash_vhash) WHERE hash=? AND vhash=? AND type=? AND uid >= ? AND (rvalue >= ? OR 0 = ?) ORDER BY uid LIMIT 1"
+#define SELECT_ENTRY_BY_HASH_VHASH_AND_TYPE "SELECT " RESULT_COLUMNS " "\
+  "FROM gn090 "\
+  "FORCE INDEX (idx_hash_vhash) "\
+  "WHERE hash = ? AND "\
+  "vhash = ? AND "\
+  "type = ? AND "\
+  "uid >= ? AND "\
+  "(rvalue >= ? OR 0 = ?) "\
+  "ORDER BY uid LIMIT 1"
   struct GNUNET_MYSQL_StatementHandle *select_entry_by_hash_vhash_and_type;
 
-#define UPDATE_ENTRY "UPDATE gn090 SET prio=prio+?,expire=IF(expire>=?,expire,?) WHERE uid=?"
+#define UPDATE_ENTRY "UPDATE gn090 SET "\
+  "prio = prio + ?, "\
+  "repl = repl + ?, "\
+  "expire = IF(expire >= ?, expire, ?) "\
+  "WHERE uid = ?"
   struct GNUNET_MYSQL_StatementHandle *update_entry;
 
 #define DEC_REPL "UPDATE gn090 SET repl=GREATEST (1, repl) - 1 WHERE uid=?"
@@ -174,20 +208,27 @@ struct Plugin
 #define SELECT_SIZE "SELECT SUM(LENGTH(value)+256) FROM gn090"
   struct GNUNET_MYSQL_StatementHandle *get_size;
 
-#define SELECT_IT_NON_ANONYMOUS "SELECT type,prio,anonLevel,expire,hash,value,uid "\
-   "FROM gn090 FORCE INDEX (idx_anonLevel_type_rvalue) "\
-   "WHERE anonLevel=0 AND type=? AND uid >= ? "\
-   "ORDER BY uid LIMIT 1"
+#define SELECT_IT_NON_ANONYMOUS "SELECT " RESULT_COLUMNS " FROM gn090 "\
+  "FORCE INDEX (idx_anonLevel_type_rvalue) "\
+  "WHERE anonLevel=0 AND "\
+  "type=? AND "\
+  "uid >= ? "\
+  "ORDER BY uid LIMIT 1"
   struct GNUNET_MYSQL_StatementHandle *zero_iter;
 
-#define SELECT_IT_EXPIRATION "SELECT type,prio,anonLevel,expire,hash,value,uid FROM gn090 FORCE INDEX (idx_expire) WHERE expire < ? ORDER BY expire ASC LIMIT 1"
+#define SELECT_IT_EXPIRATION "SELECT " RESULT_COLUMNS " FROM gn090 "\
+  "FORCE INDEX (idx_expire) "\
+  "WHERE expire < ? "\
+  "ORDER BY expire ASC LIMIT 1"
   struct GNUNET_MYSQL_StatementHandle *select_expiration;
 
-#define SELECT_IT_PRIORITY "SELECT type,prio,anonLevel,expire,hash,value,uid FROM gn090 FORCE INDEX (idx_prio) ORDER BY prio ASC LIMIT 1"
+#define SELECT_IT_PRIORITY "SELECT " RESULT_COLUMNS " FROM gn090 "\
+  "FORCE INDEX (idx_prio) "\
+  "ORDER BY prio ASC LIMIT 1"
   struct GNUNET_MYSQL_StatementHandle *select_priority;
 
-#define SELECT_IT_REPLICATION "SELECT type,prio,anonLevel,expire,hash,value,uid "\
-  "FROM gn090 FORCE INDEX (idx_repl_rvalue) "\
+#define SELECT_IT_REPLICATION "SELECT " RESULT_COLUMNS " FROM gn090 "\
+  "FORCE INDEX (idx_repl_rvalue) "\
   "WHERE repl=? AND "\
   " (rvalue>=? OR"\
   "  NOT EXISTS (SELECT 1 FROM gn090 FORCE INDEX (idx_repl_rvalue) WHERE repl=? AND rvalue>=?)) "\
@@ -371,19 +412,17 @@ mysql_plugin_put (void *cls,
 
 
 /**
- * Update the priority for a particular key in the datastore.  If
- * the expiration time in value is different than the time found in
- * the datastore, the higher value should be kept.  For the
- * anonymity level, the lower value is to be used.  The specified
- * priority should be added to the existing priority, ignoring the
- * priority in value.
- *
- * Note that it is possible for multiple values to match this put.
- * In that case, all of the respective values are updated.
+ * Update the priority, replication and expiration for a particular
+ * unique ID in the datastore.  If the expiration time in value is
+ * different than the time found in the datastore, the higher value
+ * should be kept.  The specified priority and replication is added
+ * to the existing value.
  *
  * @param cls our "struct Plugin*"
  * @param uid unique identifier of the datum
- * @param delta by how much should the priority
+ * @param priority by how much should the priority
+ *     change?
+ * @param replication by how much should the replication
  *     change?
  * @param expire new expiration time should be the
  *     MAX of any existing expiration time and
@@ -394,7 +433,8 @@ mysql_plugin_put (void *cls,
 static void
 mysql_plugin_update (void *cls,
                      uint64_t uid,
-                     uint32_t delta,
+                     uint32_t priority,
+                     uint32_t replication,
                      struct GNUNET_TIME_Absolute expire,
                      PluginUpdateCont cont,
                      void *cont_cls)
@@ -404,13 +444,15 @@ mysql_plugin_update (void *cls,
   int ret;
 
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-              "Updating value %llu adding %d to priority and maxing exp at %s\n",
+              "Updating value %llu adding %d to priority %d to replication and maxing exp at %s\n",
               (unsigned long long) uid,
-              delta,
-              GNUNET_STRINGS_absolute_time_to_string (expire));
+              priority,
+              replication,
+              GNUNET_STRINGS_absolute_time_to_string (expire));
 
   struct GNUNET_MY_QueryParam params_update[] = {
-    GNUNET_MY_query_param_uint32 (&delta),
+    GNUNET_MY_query_param_uint32 (&priority),
+    GNUNET_MY_query_param_uint32 (&replication),
     GNUNET_MY_query_param_uint64 (&lexpire),
     GNUNET_MY_query_param_uint64 (&lexpire),
     GNUNET_MY_query_param_uint64 (&uid),
@@ -457,6 +499,7 @@ execute_select (struct Plugin *plugin,
                 struct GNUNET_MY_QueryParam *params_select)
 {
   int ret;
+  uint32_t replication;
   uint32_t type;
   uint32_t priority;
   uint32_t anonymity;
@@ -466,6 +509,7 @@ execute_select (struct Plugin *plugin,
   struct GNUNET_HashCode key;
   struct GNUNET_TIME_Absolute expiration;
   struct GNUNET_MY_ResultSpec results_select[] = {
+    GNUNET_MY_result_spec_uint32 (&replication),
     GNUNET_MY_result_spec_uint32 (&type),
     GNUNET_MY_result_spec_uint32 (&priority),
     GNUNET_MY_result_spec_uint32 (&anonymity),
@@ -482,7 +526,7 @@ execute_select (struct Plugin *plugin,
   if (GNUNET_OK != ret)
   {
     proc (proc_cls,
-          NULL, 0, NULL, 0, 0, 0, GNUNET_TIME_UNIT_ZERO_ABS, 0);
+          NULL, 0, NULL, 0, 0, 0, 0, GNUNET_TIME_UNIT_ZERO_ABS, 0);
     return;
   }
 
@@ -491,7 +535,7 @@ execute_select (struct Plugin *plugin,
   if (GNUNET_OK != ret)
   {
     proc (proc_cls,
-          NULL, 0, NULL, 0, 0, 0, GNUNET_TIME_UNIT_ZERO_ABS, 0);
+          NULL, 0, NULL, 0, 0, 0, 0, GNUNET_TIME_UNIT_ZERO_ABS, 0);
     return;
   }
 
@@ -513,6 +557,7 @@ execute_select (struct Plugin *plugin,
               type,
               priority,
               anonymity,
+              replication,
               expiration,
               uid);
   GNUNET_MY_cleanup_result (results_select);
@@ -729,6 +774,7 @@ struct ReplCtx
  * @param type type of the content
  * @param priority priority of the content
  * @param anonymity anonymity-level for the content
+ * @param replication replication-level for the content
  * @param expiration expiration time for the content
  * @param uid unique identifier for the datum;
  *        maybe 0 if no unique identifier is available
@@ -744,6 +790,7 @@ repl_proc (void *cls,
            enum GNUNET_BLOCK_Type type,
            uint32_t priority,
            uint32_t anonymity,
+           uint32_t replication,
            struct GNUNET_TIME_Absolute expiration,
            uint64_t uid)
 {
@@ -759,6 +806,7 @@ repl_proc (void *cls,
                   type,
                   priority,
                   anonymity,
+                  replication,
                   expiration,
                   uid);
   if (NULL != key)
@@ -826,7 +874,7 @@ mysql_plugin_get_replication (void *cls,
                                plugin->max_repl,
                                params_get))
   {
-    proc (proc_cls, NULL, 0, NULL, 0, 0, 0, GNUNET_TIME_UNIT_ZERO_ABS, 0);
+    proc (proc_cls, NULL, 0, NULL, 0, 0, 0, 0, GNUNET_TIME_UNIT_ZERO_ABS, 0);
     return;
   }
 
@@ -834,7 +882,7 @@ mysql_plugin_get_replication (void *cls,
       GNUNET_MY_extract_result (plugin->max_repl,
                                 results_get))
   {
-    proc (proc_cls, NULL, 0, NULL, 0, 0, 0, GNUNET_TIME_UNIT_ZERO_ABS, 0);
+    proc (proc_cls, NULL, 0, NULL, 0, 0, 0, 0, GNUNET_TIME_UNIT_ZERO_ABS, 0);
     return;
   }
   GNUNET_break (GNUNET_NO ==
@@ -976,6 +1024,7 @@ struct ExpiCtx
  * @param type type of the content
  * @param priority priority of the content
  * @param anonymity anonymity-level for the content
+ * @param replication replication-level for the content
  * @param expiration expiration time for the content
  * @param uid unique identifier for the datum;
  *        maybe 0 if no unique identifier is available
@@ -991,6 +1040,7 @@ expi_proc (void *cls,
            enum GNUNET_BLOCK_Type type,
            uint32_t priority,
            uint32_t anonymity,
+           uint32_t replication,
            struct GNUNET_TIME_Absolute expiration,
            uint64_t uid)
 {
@@ -1016,6 +1066,7 @@ expi_proc (void *cls,
                    type,
                    priority,
                    anonymity,
+                   replication,
                    expiration,
                    uid);
 }