#include "gnunet_util_lib.h"
#include "gnunet_protocols.h"
#include "gnunet_statistics_service.h"
-#include "plugin_datastore.h"
+#include "gnunet_datastore_plugin.h"
#include "datastore.h"
/**
*/
#define MAX_EXPIRE_DELAY GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MINUTES, 15)
+#define QUOTA_STAT_NAME gettext_noop ("# bytes used in file-sharing datastore")
+
+/**
+ * After how many payload-changing operations
+ * do we sync our statistics?
+ */
+#define MAX_STAT_SYNC_LAG 50
/**
};
+
/**
* Our datastore plugin (NULL if not available).
*/
* How much space have we currently reserved?
*/
static unsigned long long reserved;
+
+/**
+ * How much data are we currently storing
+ * in the database?
+ */
+static unsigned long long payload;
+
+/**
+ * Number of updates that were made to the
+ * payload value since we last synchronized
+ * it with the statistics service.
+ */
+static unsigned int lastSync;
+
+/**
+ * Did we get an answer from statistics?
+ */
+static int stats_worked;
/**
* Identity of the task that is used to delete
*/
const struct GNUNET_CONFIGURATION_Handle *cfg;
-/**
- * Our scheduler.
- */
-struct GNUNET_SCHEDULER_Handle *sched;
/**
* Handle for reporting statistics.
static struct GNUNET_STATISTICS_Handle *stats;
+/**
+ * Synchronize our utilization statistics with the
+ * statistics service.
+ */
+static void
+sync_stats ()
+{
+ GNUNET_STATISTICS_set (stats,
+ QUOTA_STAT_NAME,
+ payload,
+ GNUNET_YES);
+ lastSync = 0;
+}
+
+
+
+
/**
* Function called once the transmit operation has
* either failed or succeeded.
*/
static int cleaning_done;
+/**
+ * Handle for pending get request.
+ */
+static struct GNUNET_STATISTICS_GetHandle *stat_get;
+
+
/**
* Task that is used to remove expired entries from
* the datastore. This task will schedule itself
if (key == NULL)
{
expired_kill_task
- = GNUNET_SCHEDULER_add_delayed (sched,
- MAX_EXPIRE_DELAY,
+ = GNUNET_SCHEDULER_add_delayed (MAX_EXPIRE_DELAY,
&delete_expired,
NULL);
return GNUNET_SYSERR;
}
now = GNUNET_TIME_absolute_get ();
- if (expiration.value > now.value)
+ if (expiration.abs_value > now.abs_value)
{
/* finished processing */
plugin->api->next_request (next_cls, GNUNET_YES);
"Deleting content `%s' of type %u that expired %llu ms ago\n",
GNUNET_h2s (key),
type,
- (unsigned long long) (now.value - expiration.value));
+ (unsigned long long) (now.abs_value - expiration.abs_value));
#endif
GNUNET_STATISTICS_update (stats,
gettext_noop ("# bytes expired"),
msize = ntohs(tcc->msg->size);
if (size == 0)
{
-#if DEBUG_DATASTORE
GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
- "Transmission failed.\n");
-#endif
+ _("Transmission to client failed!\n"));
if (tcc->tc != NULL)
tcc->tc (tcc->tc_cls, GNUNET_SYSERR);
if (GNUNET_YES == tcc->end)
- GNUNET_SERVER_receive_done (tcc->client, GNUNET_SYSERR);
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+ _("Disconnecting client due to transmission failure!\n"));
+ GNUNET_SERVER_receive_done (tcc->client, GNUNET_SYSERR);
+ }
GNUNET_SERVER_client_drop (tcc->client);
GNUNET_free (tcc->msg);
GNUNET_free (tcc);
GNUNET_break (0);
if (GNUNET_YES == end)
{
-#if DEBUG_DATASTORE
GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
- "Disconnecting client.\n");
-#endif
+ _("Forcefully disconnecting client.\n"));
GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
}
if (NULL != tc)
#endif
amount = GNUNET_ntohll(msg->amount);
entries = ntohl(msg->entries);
- used = plugin->api->get_size (plugin->api->cls) + reserved;
+ used = payload + reserved;
req = amount + ((unsigned long long) GNUNET_DATASTORE_ENTRY_OVERHEAD) * entries;
if (used + req > quota)
{
(GNUNET_SYSERR == ret) ? GNUNET_SYSERR : GNUNET_OK,
msg);
GNUNET_free_non_null (msg);
- if (quota - reserved - cache_size < plugin->api->get_size (plugin->api->cls))
+ if (quota - reserved - cache_size < payload)
{
GNUNET_log (GNUNET_ERROR_TYPE_INFO,
_("Need %llu bytes more space (%llu allowed, using %llu)\n"),
(unsigned long long) size + GNUNET_DATASTORE_ENTRY_OVERHEAD,
(unsigned long long) (quota - reserved - cache_size),
- (unsigned long long) plugin->api->get_size (plugin->api->cls));
+ (unsigned long long) payload);
manage_space (size + GNUNET_DATASTORE_ENTRY_OVERHEAD);
}
}
GNUNET_free (pc);
return GNUNET_SYSERR;
}
- if ( (size == ntohl(dm->size)) &&
- (0 == memcmp (&dm[1],
- data,
- size)) )
+ if ( (GNUNET_BLOCK_TYPE_FS_DBLOCK == type) ||
+ (GNUNET_BLOCK_TYPE_FS_IBLOCK == type) ||
+ ( (size == ntohl(dm->size)) &&
+ (0 == memcmp (&dm[1],
+ data,
+ size)) ) )
{
pc->is_present = GNUNET_YES;
plugin->api->next_request (next_cls, GNUNET_YES);
GNUNET_NO);
GNUNET_SERVER_client_keep (client);
plugin->api->iter_migration_order (plugin->api->cls,
- 0,
+ GNUNET_BLOCK_TYPE_ANY,
&transmit_item,
client);
}
+/**
+ * Handle GET_ZERO_ANONYMITY-message.
+ *
+ * @param cls closure
+ * @param client identification of the client
+ * @param message the actual message
+ */
+static void
+handle_get_zero_anonymity (void *cls,
+ struct GNUNET_SERVER_Client *client,
+ const struct GNUNET_MessageHeader *message)
+{
+ const struct GetZeroAnonymityMessage * msg = (const struct GetZeroAnonymityMessage*) message;
+ enum GNUNET_BLOCK_Type type;
+
+ type = (enum GNUNET_BLOCK_Type) ntohl (msg->type);
+#if DEBUG_DATASTORE
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Processing `%s' request\n",
+ "GET_ZERO_ANONYMITY");
+#endif
+ GNUNET_STATISTICS_update (stats,
+ gettext_noop ("# GET ZERO ANONYMITY requests received"),
+ 1,
+ GNUNET_NO);
+ GNUNET_SERVER_client_keep (client);
+ plugin->api->iter_zero_anonymity (plugin->api->cls,
+ type,
+ &transmit_item,
+ client);
+}
+
/**
* Context for the 'remove_callback'.
plugin->api->get (plugin->api->cls,
&dm->key,
&vhash,
- ntohl(dm->type),
+ (enum GNUNET_BLOCK_Type) ntohl(dm->type),
&remove_callback,
rc);
}
}
+/**
+ * Function called by plugins to notify us about a
+ * change in their disk utilization.
+ *
+ * @param cls closure (NULL)
+ * @param delta change in disk utilization,
+ * 0 for "reset to empty"
+ */
+static void
+disk_utilization_change_cb (void *cls,
+ int delta)
+{
+ if ( (delta < 0) &&
+ (payload < -delta) )
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+ _("Datastore payload inaccurate (%lld < %lld). Trying to fix.\n"),
+ (long long) payload,
+ (long long) -delta);
+ payload = plugin->api->get_size (plugin->api->cls);
+ sync_stats ();
+ return;
+ }
+ payload += delta;
+ lastSync++;
+ if (lastSync >= MAX_STAT_SYNC_LAG)
+ sync_stats ();
+}
+
+
+/**
+ * Callback function to process statistic values.
+ *
+ * @param cls closure (struct Plugin*)
+ * @param subsystem name of subsystem that created the statistic
+ * @param name the name of the datum
+ * @param value the current value
+ * @param is_persistent GNUNET_YES if the value is persistent, GNUNET_NO if not
+ * @return GNUNET_OK to continue, GNUNET_SYSERR to abort iteration
+ */
+static int
+process_stat_in (void *cls,
+ const char *subsystem,
+ const char *name,
+ uint64_t value,
+ int is_persistent)
+{
+ GNUNET_assert (stats_worked == GNUNET_NO);
+ stats_worked = GNUNET_YES;
+ payload += value;
+#if DEBUG_SQLITE
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Notification from statistics about existing payload (%llu), new payload is %llu\n",
+ abs_value,
+ payload);
+#endif
+ return GNUNET_OK;
+}
+
+
+static void
+process_stat_done (void *cls,
+ int success)
+{
+ struct DatastorePlugin *plugin = cls;
+
+ stat_get = NULL;
+ if (stats_worked == GNUNET_NO)
+ payload = plugin->api->get_size (plugin->api->cls);
+}
+
+
/**
* Load the datastore plugin.
*/
}
ret = GNUNET_malloc (sizeof(struct DatastorePlugin));
ret->env.cfg = cfg;
- ret->env.sched = sched;
+ ret->env.duc = &disk_utilization_change_cb;
+ ret->env.cls = NULL;
GNUNET_log (GNUNET_ERROR_TYPE_INFO,
_("Loading `%s' datastore plugin\n"), name);
GNUNET_asprintf (&libname, "libgnunet_plugin_datastore_%s", name);
GNUNET_CONTAINER_bloomfilter_free (filter);
filter = NULL;
}
+ if (lastSync > 0)
+ sync_stats ();
+ if (stat_get != NULL)
+ {
+ GNUNET_STATISTICS_get_cancel (stat_get);
+ stat_get = NULL;
+ }
if (stats != NULL)
{
GNUNET_STATISTICS_destroy (stats, GNUNET_YES);
}
if (expired_kill_task != GNUNET_SCHEDULER_NO_TASK)
{
- GNUNET_SCHEDULER_cancel (sched,
- expired_kill_task);
+ GNUNET_SCHEDULER_cancel (expired_kill_task);
expired_kill_task = GNUNET_SCHEDULER_NO_TASK;
}
- GNUNET_SCHEDULER_add_continuation (sched,
- &unload_task,
+ GNUNET_SCHEDULER_add_continuation (&unload_task,
NULL,
GNUNET_SCHEDULER_REASON_PREREQ_DONE);
}
* Process datastore requests.
*
* @param cls closure
- * @param s scheduler to use
* @param server the initialized server
* @param c configuration to use
*/
static void
run (void *cls,
- struct GNUNET_SCHEDULER_Handle *s,
struct GNUNET_SERVER_Handle *server,
const struct GNUNET_CONFIGURATION_Handle *c)
{
{&handle_get, NULL, GNUNET_MESSAGE_TYPE_DATASTORE_GET, 0 },
{&handle_get_random, NULL, GNUNET_MESSAGE_TYPE_DATASTORE_GET_RANDOM,
sizeof(struct GNUNET_MessageHeader) },
+ {&handle_get_zero_anonymity, NULL, GNUNET_MESSAGE_TYPE_DATASTORE_GET_ZERO_ANONYMITY,
+ sizeof(struct GetZeroAnonymityMessage) },
{&handle_remove, NULL, GNUNET_MESSAGE_TYPE_DATASTORE_REMOVE, 0 },
{&handle_drop, NULL, GNUNET_MESSAGE_TYPE_DATASTORE_DROP,
sizeof(struct GNUNET_MessageHeader) },
char *fn;
unsigned int bf_size;
- sched = s;
cfg = c;
if (GNUNET_OK !=
GNUNET_CONFIGURATION_get_value_number (cfg,
"DATASTORE");
return;
}
- stats = GNUNET_STATISTICS_create (sched, "datastore", cfg);
+ stats = GNUNET_STATISTICS_create ("datastore", cfg);
GNUNET_STATISTICS_set (stats,
gettext_noop ("# quota"),
quota,
}
return;
}
+ stat_get = GNUNET_STATISTICS_get (stats,
+ "datastore",
+ QUOTA_STAT_NAME,
+ GNUNET_TIME_UNIT_SECONDS,
+ &process_stat_done,
+ &process_stat_in,
+ plugin);
GNUNET_SERVER_disconnect_notify (server, &cleanup_reservations, NULL);
GNUNET_SERVER_add_handlers (server, handlers);
expired_kill_task
- = GNUNET_SCHEDULER_add_with_priority (sched,
- GNUNET_SCHEDULER_PRIORITY_IDLE,
+ = GNUNET_SCHEDULER_add_with_priority (GNUNET_SCHEDULER_PRIORITY_IDLE,
&delete_expired, NULL);
- GNUNET_SCHEDULER_add_delayed (sched,
- GNUNET_TIME_UNIT_FOREVER_REL,
+ GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_FOREVER_REL,
&cleaning_task, NULL);
-
}
projects / oweals / gnunet.git / blobdiff