X-Git-Url: https://git.librecmc.org/?a=blobdiff_plain;f=src%2Fdatastore%2Fgnunet-service-datastore.c;h=49b9db8590b65c699468760335211e387be22b97;hb=0c933ea27970e61d660a8058aa7a473df44ab029;hp=8555fad3aeb0c7e3acec35ee97d3d387f736c52c;hpb=f04c8750cca2dcfd35aceae3379b83637b8aadfc;p=oweals%2Fgnunet.git diff --git a/src/datastore/gnunet-service-datastore.c b/src/datastore/gnunet-service-datastore.c index 8555fad3a..49b9db859 100644 --- a/src/datastore/gnunet-service-datastore.c +++ b/src/datastore/gnunet-service-datastore.c @@ -48,8 +48,10 @@ */ #define MIN_EXPIRE_DELAY GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 1) - -#define QUOTA_STAT_NAME gettext_noop ("# bytes used in file-sharing datastore") +/** + * Name under which we store current space consumption. + */ +static char *quota_stat_name; /** * After how many payload-changing operations @@ -149,6 +151,11 @@ static unsigned long long quota; */ static int do_drop; +/** + * Name of our plugin. + */ +static char *plugin_name; + /** * How much space are we using for the cache? (space available for * insertions that will be instantly reclaimed by discarding less @@ -191,6 +198,13 @@ static GNUNET_SCHEDULER_TaskIdentifier expired_kill_task; */ const struct GNUNET_CONFIGURATION_Handle *cfg; +/** + * Minimum time that content should have to not be discarded instantly + * (time stamp of any content that we've been discarding recently to + * stay below the quota). FOREVER if we had to expire content with + * non-zero priority. + */ +static struct GNUNET_TIME_Absolute min_expiration; /** * Handle for reporting statistics. @@ -205,7 +219,8 @@ static struct GNUNET_STATISTICS_Handle *stats; static void sync_stats () { - GNUNET_STATISTICS_set (stats, QUOTA_STAT_NAME, payload, GNUNET_YES); + GNUNET_STATISTICS_set (stats, quota_stat_name, payload, GNUNET_YES); + GNUNET_STATISTICS_set (stats, "# utilization by current datastore", payload, GNUNET_NO); lastSync = 0; } @@ -235,7 +250,7 @@ struct TransmitCallbackContext /** * Handle for the transmission request. */ - struct GNUNET_CONNECTION_TransmitHandle *th; + struct GNUNET_SERVER_TransmitHandle *th; /** * Client that we are transmitting to. @@ -311,7 +326,9 @@ expired_processor (void *cls, const GNUNET_HashCode * key, uint32_t size, if (key == NULL) { expired_kill_task = - GNUNET_SCHEDULER_add_delayed (MAX_EXPIRE_DELAY, &delete_expired, NULL); + GNUNET_SCHEDULER_add_delayed_with_priority (MAX_EXPIRE_DELAY, + GNUNET_SCHEDULER_PRIORITY_IDLE, + &delete_expired, NULL); return GNUNET_SYSERR; } now = GNUNET_TIME_absolute_get (); @@ -319,20 +336,23 @@ expired_processor (void *cls, const GNUNET_HashCode * key, uint32_t size, { /* finished processing */ expired_kill_task = - GNUNET_SCHEDULER_add_delayed (MAX_EXPIRE_DELAY, &delete_expired, NULL); + GNUNET_SCHEDULER_add_delayed_with_priority (MAX_EXPIRE_DELAY, + GNUNET_SCHEDULER_PRIORITY_IDLE, + &delete_expired, NULL); return GNUNET_SYSERR; } -#if DEBUG_DATASTORE GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Deleting content `%s' of type %u that expired %llu ms ago\n", GNUNET_h2s (key), type, (unsigned long long) (now.abs_value - expiration.abs_value)); -#endif + min_expiration = now; GNUNET_STATISTICS_update (stats, gettext_noop ("# bytes expired"), size, GNUNET_YES); GNUNET_CONTAINER_bloomfilter_remove (filter, key); expired_kill_task = - GNUNET_SCHEDULER_add_delayed (MIN_EXPIRE_DELAY, &delete_expired, NULL); + GNUNET_SCHEDULER_add_delayed_with_priority (MIN_EXPIRE_DELAY, + GNUNET_SCHEDULER_PRIORITY_IDLE, + &delete_expired, NULL); return GNUNET_NO; } @@ -383,16 +403,21 @@ quota_processor (void *cls, const GNUNET_HashCode * key, uint32_t size, if (NULL == key) return GNUNET_SYSERR; -#if DEBUG_DATASTORE GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, - "Deleting %llu bytes of low-priority content `%s' of type %u (still trying to free another %llu bytes)\n", + "Deleting %llu bytes of low-priority (%u) content `%s' of type %u at %llu ms prior to expiration (still trying to free another %llu bytes)\n", (unsigned long long) (size + GNUNET_DATASTORE_ENTRY_OVERHEAD), - GNUNET_h2s (key), type, *need); -#endif + (unsigned int) priority, + GNUNET_h2s (key), type, + (unsigned long long) GNUNET_TIME_absolute_get_remaining (expiration).rel_value, + *need); if (size + GNUNET_DATASTORE_ENTRY_OVERHEAD > *need) *need = 0; else *need -= size + GNUNET_DATASTORE_ENTRY_OVERHEAD; + if (priority > 0) + min_expiration = GNUNET_TIME_UNIT_FOREVER_ABS; + else + min_expiration = expiration; GNUNET_STATISTICS_update (stats, gettext_noop ("# bytes purged (low-priority)"), size, GNUNET_YES); @@ -418,10 +443,8 @@ manage_space (unsigned long long need) { unsigned long long last; -#if DEBUG_DATASTORE GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Asked to free up %llu bytes of cache space\n", need); -#endif last = 0; while ((need > 0) && (last != need)) { @@ -484,10 +507,8 @@ transmit (struct GNUNET_SERVER_Client *client, struct GNUNET_MessageHeader *msg) if (GNUNET_YES == cleaning_done) { -#if DEBUG_DATASTORE GNUNET_log (GNUNET_ERROR_TYPE_WARNING, - "Shutdown in progress, aborting transmission.\n"); -#endif + _("Shutdown in progress, aborting transmission.\n")); GNUNET_SERVER_receive_done (client, GNUNET_SYSERR); GNUNET_free (msg); return; @@ -525,16 +546,15 @@ transmit_status (struct GNUNET_SERVER_Client *client, int code, const char *msg) struct StatusMessage *sm; size_t slen; -#if DEBUG_DATASTORE GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Transmitting `%s' message with value %d and message `%s'\n", "STATUS", code, msg != NULL ? msg : "(none)"); -#endif slen = (msg == NULL) ? 0 : strlen (msg) + 1; sm = GNUNET_malloc (sizeof (struct StatusMessage) + slen); sm->header.size = htons (sizeof (struct StatusMessage) + slen); sm->header.type = htons (GNUNET_MESSAGE_TYPE_DATASTORE_STATUS); sm->status = htonl (code); + sm->min_expiration = GNUNET_TIME_absolute_hton (min_expiration); if (slen > 0) memcpy (&sm[1], msg, slen); transmit (client, &sm->header); @@ -573,10 +593,8 @@ transmit_item (void *cls, const GNUNET_HashCode * key, uint32_t size, if (key == NULL) { /* transmit 'DATA_END' */ -#if DEBUG_DATASTORE GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Transmitting `%s' message\n", "DATA_END"); -#endif end = GNUNET_malloc (sizeof (struct GNUNET_MessageHeader)); end->size = htons (sizeof (struct GNUNET_MessageHeader)); end->type = htons (GNUNET_MESSAGE_TYPE_DATASTORE_DATA_END); @@ -600,13 +618,11 @@ transmit_item (void *cls, const GNUNET_HashCode * key, uint32_t size, dm->uid = GNUNET_htonll (uid); dm->key = *key; memcpy (&dm[1], data, size); -#if DEBUG_DATASTORE GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Transmitting `%s' message for `%s' of type %u with expiration %llu (now: %llu)\n", "DATA", GNUNET_h2s (key), type, (unsigned long long) expiration.abs_value, (unsigned long long) GNUNET_TIME_absolute_get ().abs_value); -#endif GNUNET_STATISTICS_update (stats, gettext_noop ("# results found"), 1, GNUNET_NO); transmit (client, &dm->header); @@ -638,9 +654,7 @@ handle_reserve (void *cls, struct GNUNET_SERVER_Client *client, uint64_t amount; uint32_t entries; -#if DEBUG_DATASTORE GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Processing `%s' request\n", "RESERVE"); -#endif amount = GNUNET_ntohll (msg->amount); entries = ntohl (msg->entries); used = payload + reserved; @@ -712,10 +726,8 @@ handle_release_reserve (void *cls, struct GNUNET_SERVER_Client *client, int rid = ntohl (msg->rid); unsigned long long rem; -#if DEBUG_DATASTORE GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Processing `%s' request\n", "RELEASE_RESERVE"); -#endif next = reservations; prev = NULL; while (NULL != (pos = next)) @@ -734,11 +746,9 @@ handle_release_reserve (void *cls, struct GNUNET_SERVER_Client *client, reserved -= rem; GNUNET_STATISTICS_set (stats, gettext_noop ("# reserved"), reserved, GNUNET_NO); -#if DEBUG_DATASTORE GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Returning %llu remaining reserved bytes to storage pool\n", rem); -#endif GNUNET_free (pos); transmit_status (client, GNUNET_OK, NULL); return; @@ -824,11 +834,9 @@ execute_put (struct GNUNET_SERVER_Client *client, const struct DataMessage *dm) GNUNET_STATISTICS_update (stats, gettext_noop ("# bytes stored"), size, GNUNET_YES); GNUNET_CONTAINER_bloomfilter_add (filter, &dm->key); -#if DEBUG_DATASTORE GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Successfully stored %u bytes of type %u under key `%s'\n", size, ntohl (dm->type), GNUNET_h2s (&dm->key)); -#endif } transmit_status (client, ret, msg); GNUNET_free_non_null (msg); @@ -884,10 +892,8 @@ check_present (void *cls, const GNUNET_HashCode * key, uint32_t size, (0 == memcmp (&dm[1], data, size)))) { -#if DEBUG_MYSQL GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Result already present in datastore\n"); -#endif /* FIXME: change API to allow increasing 'replication' counter */ if ((ntohl (dm->priority) > 0) || (GNUNET_TIME_absolute_ntoh (dm->expiration).abs_value > @@ -933,11 +939,9 @@ handle_put (void *cls, struct GNUNET_SERVER_Client *client, GNUNET_SERVER_receive_done (client, GNUNET_SYSERR); return; } -#if DEBUG_DATASTORE GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Processing `%s' request for `%s' of type %u\n", "PUT", GNUNET_h2s (&dm->key), ntohl (dm->type)); -#endif rid = ntohl (dm->rid); size = ntohl (dm->size); if (rid > 0) @@ -996,11 +1000,9 @@ handle_get (void *cls, struct GNUNET_SERVER_Client *client, return; } msg = (const struct GetMessage *) message; -#if DEBUG_DATASTORE GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Processing `%s' request for `%s' of type %u\n", "GET", GNUNET_h2s (&msg->key), ntohl (msg->type)); -#endif GNUNET_STATISTICS_update (stats, gettext_noop ("# GET requests received"), 1, GNUNET_NO); GNUNET_SERVER_client_keep (client); @@ -1008,11 +1010,9 @@ handle_get (void *cls, struct GNUNET_SERVER_Client *client, (GNUNET_YES != GNUNET_CONTAINER_bloomfilter_test (filter, &msg->key))) { /* don't bother database... */ -#if DEBUG_DATASTORE GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Empty result set for `%s' request for `%s' (bloomfilter).\n", "GET", GNUNET_h2s (&msg->key)); -#endif GNUNET_STATISTICS_update (stats, gettext_noop ("# requests filtered by bloomfilter"), 1, @@ -1047,10 +1047,8 @@ handle_update (void *cls, struct GNUNET_SERVER_Client *client, 1, GNUNET_NO); msg = (const struct UpdateMessage *) message; emsg = NULL; -#if DEBUG_DATASTORE GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Processing `%s' request for %llu\n", "UPDATE", (unsigned long long) GNUNET_ntohll (msg->uid)); -#endif ret = plugin->api->update (plugin->api->cls, GNUNET_ntohll (msg->uid), (int32_t) ntohl (msg->priority), @@ -1071,10 +1069,8 @@ static void handle_get_replication (void *cls, struct GNUNET_SERVER_Client *client, const struct GNUNET_MessageHeader *message) { -#if DEBUG_DATASTORE GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Processing `%s' request\n", "GET_REPLICATION"); -#endif GNUNET_STATISTICS_update (stats, gettext_noop ("# GET REPLICATION requests received"), 1, @@ -1106,10 +1102,8 @@ handle_get_zero_anonymity (void *cls, struct GNUNET_SERVER_Client *client, GNUNET_SERVER_receive_done (client, GNUNET_SYSERR); return; } -#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, @@ -1135,19 +1129,15 @@ remove_callback (void *cls, const GNUNET_HashCode * key, uint32_t size, if (key == NULL) { -#if DEBUG_DATASTORE GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "No further matches for `%s' request.\n", "REMOVE"); -#endif transmit_status (client, GNUNET_NO, _("Content not found")); GNUNET_SERVER_client_drop (client); return GNUNET_OK; /* last item */ } -#if DEBUG_DATASTORE GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Item %llu matches `%s' request for key `%s' and type %u.\n", (unsigned long long) uid, "REMOVE", GNUNET_h2s (key), type); -#endif GNUNET_STATISTICS_update (stats, gettext_noop ("# bytes removed (explicit request)"), size, GNUNET_YES); @@ -1178,11 +1168,9 @@ handle_remove (void *cls, struct GNUNET_SERVER_Client *client, GNUNET_SERVER_receive_done (client, GNUNET_SYSERR); return; } -#if DEBUG_DATASTORE GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Processing `%s' request for `%s' of type %u\n", "REMOVE", GNUNET_h2s (&dm->key), ntohl (dm->type)); -#endif GNUNET_STATISTICS_update (stats, gettext_noop ("# REMOVE requests received"), 1, GNUNET_NO); GNUNET_SERVER_client_keep (client); @@ -1204,9 +1192,7 @@ static void handle_drop (void *cls, struct GNUNET_SERVER_Client *client, const struct GNUNET_MessageHeader *message) { -#if DEBUG_DATASTORE GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Processing `%s' request\n", "DROP"); -#endif do_drop = GNUNET_YES; GNUNET_SERVER_receive_done (client, GNUNET_OK); } @@ -1257,11 +1243,9 @@ process_stat_in (void *cls, const char *subsystem, const char *name, 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 + value, payload); return GNUNET_OK; } @@ -1285,31 +1269,21 @@ load_plugin () { struct DatastorePlugin *ret; char *libname; - char *name; - if (GNUNET_OK != - GNUNET_CONFIGURATION_get_value_string (cfg, "DATASTORE", "DATABASE", - &name)) - { - GNUNET_log (GNUNET_ERROR_TYPE_ERROR, - _("No `%s' specified for `%s' in configuration!\n"), "DATABASE", - "DATASTORE"); - return NULL; - } ret = GNUNET_malloc (sizeof (struct DatastorePlugin)); ret->env.cfg = cfg; 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); - ret->short_name = name; + plugin_name); + GNUNET_asprintf (&libname, "libgnunet_plugin_datastore_%s", plugin_name); + ret->short_name = GNUNET_strdup (plugin_name); ret->lib_name = libname; ret->api = GNUNET_PLUGIN_load (libname, &ret->env); if (ret->api == NULL) { GNUNET_log (GNUNET_ERROR_TYPE_ERROR, - _("Failed to load datastore plugin for `%s'\n"), name); + _("Failed to load datastore plugin for `%s'\n"), plugin_name); GNUNET_free (ret->short_name); GNUNET_free (libname); GNUNET_free (ret); @@ -1328,14 +1302,14 @@ load_plugin () static void unload_plugin (struct DatastorePlugin *plug) { -#if DEBUG_DATASTORE GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Datastore service is unloading plugin...\n"); -#endif GNUNET_break (NULL == GNUNET_PLUGIN_unload (plug->lib_name, plug->api)); GNUNET_free (plug->lib_name); GNUNET_free (plug->short_name); GNUNET_free (plug); + GNUNET_free (quota_stat_name); + quota_stat_name = NULL; } @@ -1346,6 +1320,8 @@ unload_plugin (struct DatastorePlugin *plug) static void unload_task (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc) { + if (lastSync > 0) + sync_stats (); if (GNUNET_YES == do_drop) plugin->api->drop (plugin->api->cls); unload_plugin (plugin); @@ -1355,8 +1331,6 @@ unload_task (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc) GNUNET_CONTAINER_bloomfilter_free (filter); filter = NULL; } - if (lastSync > 0) - sync_stats (); if (stat_get != NULL) { GNUNET_STATISTICS_get_cancel (stat_get); @@ -1367,6 +1341,8 @@ unload_task (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc) GNUNET_STATISTICS_destroy (stats, GNUNET_YES); stats = NULL; } + GNUNET_free_non_null (plugin_name); + plugin_name = NULL; } @@ -1385,7 +1361,7 @@ cleaning_task (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc) GNUNET_CONTAINER_DLL_remove (tcc_head, tcc_tail, tcc); if (tcc->th != NULL) { - GNUNET_CONNECTION_notify_transmit_ready_cancel (tcc->th); + GNUNET_SERVER_notify_transmit_ready_cancel (tcc->th); GNUNET_SERVER_client_drop (tcc->client); } GNUNET_free (tcc->msg); @@ -1493,11 +1469,24 @@ run (void *cls, struct GNUNET_SERVER_Handle *server, sizeof (struct GNUNET_MessageHeader)}, {NULL, NULL, 0, 0} }; - char *fn; + char *fn; + char *pfn; unsigned int bf_size; int refresh_bf; cfg = c; + if (GNUNET_OK != + GNUNET_CONFIGURATION_get_value_string (cfg, "DATASTORE", "DATABASE", + &plugin_name)) + { + GNUNET_log (GNUNET_ERROR_TYPE_ERROR, + _("No `%s' specified for `%s' in configuration!\n"), "DATABASE", + "DATASTORE"); + return; + } + GNUNET_asprintf ("a_stat_name, + _("# bytes used in file-sharing datastore `%s'"), + plugin_name); if (GNUNET_OK != GNUNET_CONFIGURATION_get_value_size (cfg, "DATASTORE", "QUOTA", "a)) { @@ -1511,10 +1500,10 @@ run (void *cls, struct GNUNET_SERVER_Handle *server, cache_size = quota / 8; /* Or should we make this an option? */ GNUNET_STATISTICS_set (stats, gettext_noop ("# cache size"), cache_size, GNUNET_NO); - if (quota / 32LL > (1 << 31)) + if (quota / (32 * 1024LL) > (1 << 31)) bf_size = (1 << 31); /* absolute limit: ~2 GB, beyond that BF just won't help anyway */ else - bf_size = quota / 32; /* 8 bit per entry, 1 bit per 32 kb in DB */ + bf_size = quota / (32 * 1024LL); /* 8 bit per entry, 1 bit per 32 kb in DB */ fn = NULL; if ((GNUNET_OK != GNUNET_CONFIGURATION_get_value_filename (cfg, "DATASTORE", "BLOOMFILTER", @@ -1529,36 +1518,37 @@ run (void *cls, struct GNUNET_SERVER_Handle *server, } if (fn != NULL) { - if (GNUNET_YES == GNUNET_DISK_file_test (fn)) + GNUNET_asprintf (&pfn, "%s.%s", fn, plugin_name); + if (GNUNET_YES == GNUNET_DISK_file_test (pfn)) { - filter = GNUNET_CONTAINER_bloomfilter_load (fn, bf_size, 5); /* approx. 3% false positives at max use */ + filter = GNUNET_CONTAINER_bloomfilter_load (pfn, bf_size, 5); /* approx. 3% false positives at max use */ if (NULL == filter) { /* file exists but not valid, remove and try again, but refresh */ - if (0 != UNLINK (fn)) + if (0 != UNLINK (pfn)) { /* failed to remove, run without file */ GNUNET_log (GNUNET_ERROR_TYPE_ERROR, _("Failed to remove bogus bloomfilter file `%s'\n"), - fn); - GNUNET_free (fn); - fn = NULL; + pfn); + GNUNET_free (pfn); + pfn = NULL; filter = GNUNET_CONTAINER_bloomfilter_load (NULL, bf_size, 5); /* approx. 3% false positives at max use */ refresh_bf = GNUNET_YES; } else { /* try again after remove */ - filter = GNUNET_CONTAINER_bloomfilter_load (fn, bf_size, 5); /* approx. 3% false positives at max use */ + filter = GNUNET_CONTAINER_bloomfilter_load (pfn, bf_size, 5); /* approx. 3% false positives at max use */ refresh_bf = GNUNET_YES; if (NULL == filter) { /* failed yet again, give up on using file */ GNUNET_log (GNUNET_ERROR_TYPE_ERROR, _("Failed to remove bogus bloomfilter file `%s'\n"), - fn); - GNUNET_free (fn); - fn = NULL; + pfn); + GNUNET_free (pfn); + pfn = NULL; filter = GNUNET_CONTAINER_bloomfilter_load (NULL, bf_size, 5); /* approx. 3% false positives at max use */ } } @@ -1571,9 +1561,10 @@ run (void *cls, struct GNUNET_SERVER_Handle *server, } else { - filter = GNUNET_CONTAINER_bloomfilter_load (fn, bf_size, 5); /* approx. 3% false positives at max use */ + filter = GNUNET_CONTAINER_bloomfilter_load (pfn, bf_size, 5); /* approx. 3% false positives at max use */ refresh_bf = GNUNET_YES; } + GNUNET_free (pfn); } else { @@ -1605,7 +1596,7 @@ run (void *cls, struct GNUNET_SERVER_Handle *server, return; } stat_get = - GNUNET_STATISTICS_get (stats, "datastore", QUOTA_STAT_NAME, + 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);