commit f286cd0b87731774b9448a0cdc82f7c8682c4efe
parent 7d18e9d89d2c920f9139bec7f55f01b00b5fd81a
Author: Omar Tarabai <tarabai@devegypt.com>
Date: Wed, 4 Jun 2014 16:03:17 +0000
peerstore: added 'replace' option and other fixes
Diffstat:
9 files changed, 126 insertions(+), 37 deletions(-)
diff --git a/src/include/gnunet_peerstore_plugin.h b/src/include/gnunet_peerstore_plugin.h
@@ -68,7 +68,8 @@ struct GNUNET_PEERSTORE_PluginFunctions
const char *key,
const void *value,
size_t size,
- struct GNUNET_TIME_Absolute expiry);
+ struct GNUNET_TIME_Absolute expiry,
+ enum GNUNET_PEERSTORE_StoreOption options);
/**
* Iterate over the records given an optional peer id
diff --git a/src/include/gnunet_peerstore_service.h b/src/include/gnunet_peerstore_service.h
@@ -38,6 +38,25 @@ extern "C"
#endif
/**
+ * Options for storing values in PEERSTORE
+ */
+enum GNUNET_PEERSTORE_StoreOption
+{
+
+ /**
+ * Possibly store multiple values under given key.
+ */
+ GNUNET_PEERSTORE_STOREOPTION_MULTIPLE,
+
+ /**
+ * Delete any previous values for the given key before
+ * storing the given value.
+ */
+ GNUNET_PEERSTORE_STOREOPTION_REPLACE,
+
+};
+
+/**
* Handle to the peerstore service.
*/
struct GNUNET_PEERSTORE_Handle;
@@ -130,7 +149,8 @@ GNUNET_PEERSTORE_disconnect(struct GNUNET_PEERSTORE_Handle *h);
* @param key entry key
* @param value entry value BLOB
* @param size size of 'value'
- * @param lifetime relative time after which the entry is (possibly) deleted
+ * @param expiry absolute time after which the entry is (possibly) deleted
+ * @param options store operation option
* @param cont Continuation function after the store request is processed
* @param cont_cls Closure for 'cont'
*/
@@ -142,6 +162,7 @@ GNUNET_PEERSTORE_store (struct GNUNET_PEERSTORE_Handle *h,
const void *value,
size_t size,
struct GNUNET_TIME_Absolute expiry,
+ enum GNUNET_PEERSTORE_StoreOption options,
GNUNET_PEERSTORE_Continuation cont,
void *cont_cls);
diff --git a/src/peerstore/gnunet-service-peerstore.c b/src/peerstore/gnunet-service-peerstore.c
@@ -30,24 +30,6 @@
#include "peerstore_common.h"
/**
- * Context of a PEERSTORE watch
- */
-struct WatchContext
-{
-
- /**
- * Hash of key of watched record
- */
- struct GNUNET_HashCode keyhash;
-
- /**
- * Client requested the watch
- */
- struct GNUNET_SERVER_Client *client;
-
-};
-
-/**
* Interval for expired records cleanup (in seconds)
*/
#define CLEANUP_INTERVAL 300 /* 5mins */
@@ -95,6 +77,7 @@ shutdown_task (void *cls,
}
GNUNET_SERVER_notification_context_destroy(nc);
GNUNET_CONTAINER_multihashmap_destroy(watchers);
+ watchers = NULL;
GNUNET_SCHEDULER_shutdown();
}
@@ -123,7 +106,7 @@ cleanup_expired_records(void *cls,
* @param cls closuer, a 'struct GNUNET_PEERSTORE_Record *'
* @param key hash of record key
* @param value the watcher client, a 'struct GNUNET_SERVER_Client *'
- * @return #GNUNET_YES to continue iterating
+ * @return #GNUNET_OK to continue iterating
*/
int client_disconnect_it(void *cls,
const struct GNUNET_HashCode *key,
@@ -131,7 +114,7 @@ int client_disconnect_it(void *cls,
{
if(cls == value)
GNUNET_CONTAINER_multihashmap_remove(watchers, key, value);
- return GNUNET_YES;
+ return GNUNET_OK;
}
/**
@@ -146,8 +129,9 @@ handle_client_disconnect (void *cls,
* client)
{
GNUNET_log(GNUNET_ERROR_TYPE_DEBUG, "A client was disconnected, cleaning up.\n");
- GNUNET_CONTAINER_multihashmap_iterate(watchers,
- &client_disconnect_it, client);
+ if(NULL != watchers)
+ GNUNET_CONTAINER_multihashmap_iterate(watchers,
+ &client_disconnect_it, client);
}
/**
@@ -196,12 +180,6 @@ int watch_notifier_it(void *cls,
struct StoreRecordMessage *srm;
GNUNET_log(GNUNET_ERROR_TYPE_DEBUG, "Found a watcher to update.\n");
- if(NULL == client)
- {
- GNUNET_log(GNUNET_ERROR_TYPE_DEBUG, "Removing a dead client.\n");
- GNUNET_CONTAINER_multihashmap_remove(watchers, key, client);
- return GNUNET_YES;
- }
srm = PEERSTORE_create_record_message(record->sub_system,
record->peer,
record->key,
@@ -335,6 +313,7 @@ void handle_store (void *cls,
const struct GNUNET_MessageHeader *message)
{
struct GNUNET_PEERSTORE_Record *record;
+ struct StoreRecordMessage *srm;
record = PEERSTORE_parse_record_message(message);
if(NULL == record)
@@ -343,6 +322,7 @@ void handle_store (void *cls,
GNUNET_SERVER_receive_done(client, GNUNET_SYSERR);
return;
}
+ srm = (struct StoreRecordMessage *)message;
if(NULL == record->sub_system
|| NULL == record->peer
|| NULL == record->key)
@@ -363,7 +343,8 @@ void handle_store (void *cls,
record->key,
record->value,
record->value_size,
- *record->expiry))
+ *record->expiry,
+ srm->options))
{
GNUNET_log(GNUNET_ERROR_TYPE_ERROR, "Failed to store requested value, sqlite database error.");
PEERSTORE_destroy_record(record);
diff --git a/src/peerstore/peerstore.h b/src/peerstore/peerstore.h
@@ -75,6 +75,12 @@ struct StoreRecordMessage
*/
struct GNUNET_TIME_Absolute expiry;
+ /**
+ * Options, needed only in case of a
+ * store operation
+ */
+ enum GNUNET_PEERSTORE_StoreOption options;
+
};
/**
diff --git a/src/peerstore/peerstore_api.c b/src/peerstore/peerstore_api.c
@@ -419,6 +419,7 @@ GNUNET_PEERSTORE_store (struct GNUNET_PEERSTORE_Handle *h,
const void *value,
size_t size,
struct GNUNET_TIME_Absolute expiry,
+ enum GNUNET_PEERSTORE_StoreOption options,
GNUNET_PEERSTORE_Continuation cont,
void *cont_cls)
{
@@ -434,6 +435,7 @@ GNUNET_PEERSTORE_store (struct GNUNET_PEERSTORE_Handle *h,
value,
size,
&expiry,
+ options,
GNUNET_MESSAGE_TYPE_PEERSTORE_STORE);
sc = GNUNET_new(struct GNUNET_PEERSTORE_StoreContext);
sc->ev = ev;
@@ -595,6 +597,7 @@ GNUNET_PEERSTORE_iterate (struct GNUNET_PEERSTORE_Handle *h,
NULL,
0,
NULL,
+ 0,
GNUNET_MESSAGE_TYPE_PEERSTORE_ITERATE);
ic = GNUNET_new(struct GNUNET_PEERSTORE_IterateContext);
ic->callback = callback;
diff --git a/src/peerstore/peerstore_common.c b/src/peerstore/peerstore_common.c
@@ -137,6 +137,7 @@ PEERSTORE_create_record_mq_envelope(const char *sub_system,
const void *value,
size_t value_size,
struct GNUNET_TIME_Absolute *expiry,
+ enum GNUNET_PEERSTORE_StoreOption options,
uint16_t msg_type)
{
struct StoreRecordMessage *srm;
@@ -168,6 +169,7 @@ PEERSTORE_create_record_mq_envelope(const char *sub_system,
}
srm->sub_system_size = htons(ss_size);
srm->value_size = htons(value_size);
+ srm->options = options;
dummy = &srm[1];
memcpy(dummy, sub_system, ss_size);
dummy += ss_size;
diff --git a/src/peerstore/peerstore_common.h b/src/peerstore/peerstore_common.h
@@ -76,6 +76,7 @@ PEERSTORE_create_record_mq_envelope(const char *sub_system,
const void *value,
size_t value_size,
struct GNUNET_TIME_Absolute *expiry,
+ enum GNUNET_PEERSTORE_StoreOption options,
uint16_t msg_type);
/**
diff --git a/src/peerstore/plugin_peerstore_sqlite.c b/src/peerstore/plugin_peerstore_sqlite.c
@@ -100,13 +100,58 @@ struct Plugin
sqlite3_stmt *select_peerstoredata_by_all;
/**
- * Precompiled SQL for deleting expired records from peerstoredata
+ * Precompiled SQL for deleting expired
+ * records from peerstoredata
*/
sqlite3_stmt *expire_peerstoredata;
+ /**
+ * Precompiled SQL for deleting records
+ * with given key
+ */
+ sqlite3_stmt *delete_peerstoredata;
+
};
/**
+ * Delete records with the given key
+ *
+ * @param cls closure (internal context for the plugin)
+ * @param sub_system name of sub system
+ * @param peer Peer identity (can be NULL)
+ * @param key entry key string (can be NULL)
+ * @return number of deleted records
+ */
+static int
+peerstore_sqlite_delete_records(void *cls,
+ const char *sub_system,
+ const struct GNUNET_PeerIdentity *peer,
+ const char *key)
+{
+ struct Plugin *plugin = cls;
+ sqlite3_stmt *stmt = plugin->delete_peerstoredata;
+
+ if((SQLITE_OK != sqlite3_bind_text(stmt, 1, sub_system, strlen(sub_system) + 1, SQLITE_STATIC))
+ || (SQLITE_OK != sqlite3_bind_blob(stmt, 2, peer, sizeof(struct GNUNET_PeerIdentity), SQLITE_STATIC))
+ || (SQLITE_OK != sqlite3_bind_text(stmt, 3, key, strlen(key) + 1, SQLITE_STATIC)))
+ {
+ LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK, "sqlite3_bind");
+ }
+ else if (SQLITE_DONE != sqlite3_step (stmt))
+ {
+ LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
+ "sqlite3_step");
+ }
+ if (SQLITE_OK != sqlite3_reset (stmt))
+ {
+ LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
+ "sqlite3_reset");
+ return 0;
+ }
+ return sqlite3_changes(plugin->dbh);
+}
+
+/**
* Delete expired records (expiry < now)
*
* @param cls closure (internal context for the plugin)
@@ -120,7 +165,7 @@ peerstore_sqlite_expire_records(void *cls,
struct Plugin *plugin = cls;
sqlite3_stmt *stmt = plugin->expire_peerstoredata;
- if(SQLITE_OK != sqlite3_bind_int64(stmt, 1, (sqlite3_int64)now.abs_value_us))
+ if(SQLITE_OK != sqlite3_bind_int64(stmt, 1, (sqlite3_uint64)now.abs_value_us))
{
LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK, "sqlite3_bind");
}
@@ -254,16 +299,21 @@ peerstore_sqlite_store_record(void *cls,
const char *key,
const void *value,
size_t size,
- struct GNUNET_TIME_Absolute expiry)
+ struct GNUNET_TIME_Absolute expiry,
+ enum GNUNET_PEERSTORE_StoreOption options)
{
struct Plugin *plugin = cls;
sqlite3_stmt *stmt = plugin->insert_peerstoredata;
+ if(GNUNET_PEERSTORE_STOREOPTION_REPLACE == options)
+ {
+ peerstore_sqlite_delete_records(cls, sub_system, peer, key);
+ }
if(SQLITE_OK != sqlite3_bind_text(stmt, 1, sub_system, strlen(sub_system) + 1, SQLITE_STATIC)
|| SQLITE_OK != sqlite3_bind_blob(stmt, 2, peer, sizeof(struct GNUNET_PeerIdentity), SQLITE_STATIC)
|| SQLITE_OK != sqlite3_bind_text(stmt, 3, key, strlen(key) + 1, SQLITE_STATIC)
|| SQLITE_OK != sqlite3_bind_blob(stmt, 4, value, size, SQLITE_STATIC)
- || SQLITE_OK != sqlite3_bind_int64(stmt, 5, (sqlite3_int64)expiry.abs_value_us))
+ || SQLITE_OK != sqlite3_bind_int64(stmt, 5, (sqlite3_uint64)expiry.abs_value_us))
LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
"sqlite3_bind");
else if (SQLITE_DONE != sqlite3_step (stmt))
@@ -330,6 +380,21 @@ sql_prepare (sqlite3 *dbh, const char *sql, sqlite3_stmt **stmt)
}
/**
+ * sqlite3 custom function for comparison of uint64_t values
+ * since it is not supported by default
+ */
+void sqlite3_lessthan(sqlite3_context* ctx, int dummy,
+ sqlite3_value** values)
+{
+ uint64_t v1;
+ uint64_t v2;
+
+ v1 = (uint64_t)sqlite3_value_int64(values[0]);
+ v2 = (uint64_t)sqlite3_value_int64(values[1]);
+ sqlite3_result_int(ctx, v1 < v2);
+}
+
+/**
* Initialize the database connections and associated
* data structures (create tables and indices
* as needed as well).
@@ -389,9 +454,11 @@ database_setup (struct Plugin *plugin)
" peer_id BLOB NOT NULL,\n"
" key TEXT NOT NULL,\n"
" value BLOB NULL,\n"
- " expiry INTEGER NOT NULL"
+ " expiry sqlite3_uint64 NOT NULL"
");");
+ sqlite3_create_function(plugin->dbh, "UINT64_LT", 2, SQLITE_UTF8, NULL, &sqlite3_lessthan, NULL, NULL);
+
/* Prepare statements */
sql_prepare (plugin->dbh,
@@ -419,8 +486,14 @@ database_setup (struct Plugin *plugin)
&plugin->select_peerstoredata_by_all);
sql_prepare(plugin->dbh,
"DELETE FROM peerstoredata"
- " WHERE expiry < ?",
+ " WHERE UINT64_LT(expiry, ?)",
&plugin->expire_peerstoredata);
+ sql_prepare(plugin->dbh,
+ "DELETE FROM peerstoredata"
+ " WHERE sub_system = ?"
+ " AND peer_id = ?"
+ " AND key = ?",
+ &plugin->delete_peerstoredata);
return GNUNET_OK;
}
diff --git a/src/peerstore/test_peerstore_api.c b/src/peerstore/test_peerstore_api.c
@@ -125,6 +125,7 @@ run (void *cls,
val,
val_size,
expiry,
+ GNUNET_PEERSTORE_STOREOPTION_MULTIPLE,
&store_cont,
NULL);
}
