X-Git-Url: https://git.librecmc.org/?a=blobdiff_plain;f=src%2Frps%2Fgnunet-service-rps_sampler.c;h=b51867d28b942e32b8edd392e08ec419451ef767;hb=36fd53ed6ae7c60d4b49cea614d18aa6d58843a1;hp=f80a701086b03b5185c68a92c4f79326316145c7;hpb=23660470add3cd0de11e3e599f14ec59f80ef5c3;p=oweals%2Fgnunet.git diff --git a/src/rps/gnunet-service-rps_sampler.c b/src/rps/gnunet-service-rps_sampler.c index f80a70108..b51867d28 100644 --- a/src/rps/gnunet-service-rps_sampler.c +++ b/src/rps/gnunet-service-rps_sampler.c @@ -1,6 +1,6 @@ /* This file is part of GNUnet. - (C) + Copyright (C) GNUnet is free software; you can redistribute it and/or modify it under the terms of the GNU General Public License as published @@ -14,8 +14,8 @@ You should have received a copy of the GNU General Public License along with GNUnet; see the file COPYING. If not, write to the - Free Software Foundation, Inc., 59 Temple Place - Suite 330, - Boston, MA 02111-1307, USA. + Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, + Boston, MA 02110-1301, USA. */ /** @@ -32,7 +32,10 @@ #include #include -#define LOG(kind, ...) GNUNET_log(kind, __VA_ARGS__) +#include "rps-test_util.h" + +#define LOG(kind, ...) GNUNET_log_from(kind,"rps-sampler",__VA_ARGS__) + // multiple 'clients'? @@ -77,17 +80,123 @@ struct RPS_SamplerElement */ struct GNUNET_HashCode peer_id_hash; + /** * Time of last request. */ - struct GNUNET_TIME_Absolute last_request; - + struct GNUNET_TIME_Absolute last_client_request; + /** * Flag that indicates that we are not holding a valid PeerID right now. */ enum RPS_SamplerEmpty is_empty; + + /** + * 'Birth' + */ + struct GNUNET_TIME_Absolute birth; + + /** + * How many times a PeerID was put in this sampler. + */ + uint32_t num_peers; + + /** + * How many times this sampler changed the peer_id. + */ + uint32_t num_change; + + /** + * The file name this sampler element should log to + */ + #ifdef TO_FILE + char *file_name; + #endif /* TO_FILE */ +}; + +/** + * Callback that is called from _get_rand_peer() when the PeerID is ready. + * + * @param cls the closure given alongside this function. + * @param id the PeerID that was returned + */ +typedef void +(*RPS_sampler_rand_peer_ready_cont) (void *cls, + const struct GNUNET_PeerIdentity *id); + + +/** + * Closure for #sampler_mod_get_rand_peer() and #sampler_get_rand_peer + */ +struct GetPeerCls +{ + /** + * DLL + */ + struct GetPeerCls *next; + + /** + * DLL + */ + struct GetPeerCls *prev; + + /** + * The sampler this function operates on. + */ + struct RPS_Sampler *sampler; + + /** + * The task for this function. + */ + struct GNUNET_SCHEDULER_Task *get_peer_task; + + /** + * The callback + */ + RPS_sampler_rand_peer_ready_cont cont; + + /** + * The closure to the callback @e cont + */ + void *cont_cls; + + /** + * The address of the id to be stored at + */ + struct GNUNET_PeerIdentity *id; }; + +/** + * Type of function used to differentiate between modified and not modified + * Sampler. + */ +typedef void +(*RPS_get_peers_type) (void *cls, + const struct GNUNET_SCHEDULER_TaskContext *tc); + +/** + * Get one random peer out of the sampled peers. + * + * We might want to reinitialise this sampler after giving the + * corrsponding peer to the client. + * Only used internally + */ +static void +sampler_get_rand_peer (void *cls, + const struct GNUNET_SCHEDULER_TaskContext *tc); + +/** + * Get one random peer out of the sampled peers. + * + * We might want to reinitialise this sampler after giving the + * corrsponding peer to the client. + */ +static void +sampler_mod_get_rand_peer (void *cls, + const struct GNUNET_SCHEDULER_TaskContext *tc); + + /** * Sampler with its own array of SamplerElements */ @@ -100,42 +209,95 @@ struct RPS_Sampler //size_t size; /** - * All Samplers in one array. + * All sampler elements in one array. */ struct RPS_SamplerElement **sampler_elements; /** - * Index to a sampler element. + * Number of sampler elements trash can holds. + */ + unsigned int trash_can_size; + + /** + * Trash can for old sampler elements. + * We need this to evaluate the sampler. + * TODO remove after evaluation + * and undo changes in + * sampler_resize + * sampler_empty + * sampler_init + * sampler_remove? + * sampler_reinitialise_by_value + * sampler_update + */ + struct RPS_SamplerElement **trash_can; + + /** + * Maximum time a round takes * - * Gets cycled on every hist_request. + * Used in the context of RPS */ - uint64_t sampler_elem_index; + struct GNUNET_TIME_Relative max_round_interval; /** - * Callback to be called when a peer gets inserted into a sampler. + * Stores the function to return peers. Which one it is depends on whether + * the Sampler is the modified one or not. */ - RPS_sampler_insert_cb insert_cb; + RPS_get_peers_type get_peers; /** - * Closure to the insert_cb. + * Head for the DLL to store the closures to pending requests. */ - void *insert_cls; + struct GetPeerCls *gpc_head; /** - * Callback to be called when a peer gets inserted into a sampler. + * Tail for the DLL to store the closures to pending requests. */ - RPS_sampler_remove_cb remove_cb; + struct GetPeerCls *gpc_tail; + #ifdef TO_FILE /** - * Closure to the remove_cb. + * File name to log to */ - void *remove_cls; + char *file_name; + #endif /* TO_FILE */ }; /** - * Global sampler variable. + * Closure to _get_n_rand_peers_ready_cb() */ -struct RPS_Sampler *sampler; +struct NRandPeersReadyCls +{ + /** + * Number of peers we are waiting for. + */ + uint32_t num_peers; + + /** + * Number of peers we currently have. + */ + uint32_t cur_num_peers; + + /** + * Pointer to the array holding the ids. + */ + struct GNUNET_PeerIdentity *ids; + + /** + * Callback to be called when all ids are available. + */ + RPS_sampler_n_rand_peers_ready_cb callback; + + /** + * Closure given to the callback + */ + void *cls; +}; + +///** +// * Global sampler variable. +// */ +//struct RPS_Sampler *sampler; /** @@ -151,67 +313,45 @@ static size_t max_size; /** * The size the extended sampler elements currently have. */ -static size_t extra_size; +//static size_t extra_size; /** * Inedex to the sampler element that is the next to be returned */ -static struct RPS_SamplerElement **extended_samplers_index; +static uint32_t client_get_index; -/** - * Request counter. - * - * Only needed in the beginning to check how many of the 64 deltas - * we already have - */ -static unsigned int req_counter; /** - * Time of the last request we received. + * Callback to _get_rand_peer() used by _get_n_rand_peers(). * - * Used to compute the expected request rate. - */ -static struct GNUNET_TIME_Absolute last_request; - -/** - * Last 64 deltas between requests + * Checks whether all n peers are available. If they are, + * give those back. */ -static struct GNUNET_TIME_Relative request_deltas[64]; +static void +check_n_peers_ready (void *cls, + const struct GNUNET_PeerIdentity *id) +{ + struct NRandPeersReadyCls *n_peers_cls = cls; -/** - * The prediction of the rate of requests - */ -static struct GNUNET_TIME_Relative request_rate; + n_peers_cls->cur_num_peers++; + LOG (GNUNET_ERROR_TYPE_DEBUG, + "Got %" PRIX32 ". of %" PRIX32 " peers\n", + n_peers_cls->cur_num_peers, n_peers_cls->num_peers); + if (n_peers_cls->num_peers == n_peers_cls->cur_num_peers) + { /* All peers are ready -- return those to the client */ + GNUNET_assert (NULL != n_peers_cls->callback); -/** - * Sum all time relatives of an array. - */ - struct GNUNET_TIME_Relative -T_relative_sum (const struct GNUNET_TIME_Relative *rel_array, uint64_t arr_size) -{ - struct GNUNET_TIME_Relative sum; - uint64_t i; + LOG (GNUNET_ERROR_TYPE_DEBUG, + "returning %" PRIX32 " peers to the client\n", + n_peers_cls->num_peers); + n_peers_cls->callback (n_peers_cls->cls, n_peers_cls->ids, n_peers_cls->num_peers); - sum = GNUNET_TIME_UNIT_ZERO; - for ( i = 0 ; i < arr_size ; i++ ) - { - sum = GNUNET_TIME_relative_add (sum, rel_array[i]); + GNUNET_free (n_peers_cls); } - return sum; -} - -/** - * Compute the average of given time relatives. - */ - struct GNUNET_TIME_Relative -T_relative_avg (const struct GNUNET_TIME_Relative *rel_array, uint64_t arr_size) -{ - return T_relative_sum (rel_array, arr_size); // FIXME find a way to devide that by arr_size } - /** * Reinitialise a previously initialised sampler element. * @@ -227,13 +367,20 @@ RPS_sampler_elem_reinit (struct RPS_SamplerElement *sampler_el) &(sampler_el->auth_key.key), GNUNET_CRYPTO_HASH_LENGTH); - sampler_el->last_request = GNUNET_TIME_UNIT_FOREVER_ABS; + #ifdef TO_FILE + /* Create a file(-name) to store internals to */ + char *name_buf; + name_buf = auth_key_to_string (sampler_el->auth_key); + + sampler_el->file_name = create_file (name_buf); + GNUNET_free (name_buf); + #endif /* TO_FILE */ - /* We might want to keep the previous peer */ + sampler_el->last_client_request = GNUNET_TIME_UNIT_FOREVER_ABS; - //GNUNET_CRYPTO_hmac(&sampler_el->auth_key, sampler_el->peer_id, - // sizeof(struct GNUNET_PeerIdentity), - // &sampler_el->peer_id_hash); + sampler_el->birth = GNUNET_TIME_absolute_get (); + sampler_el->num_peers = 0; + sampler_el->num_change = 0; } @@ -249,32 +396,40 @@ RPS_sampler_elem_reinit (struct RPS_SamplerElement *sampler_el) RPS_sampler_elem_create (void) { struct RPS_SamplerElement *s; - + s = GNUNET_new (struct RPS_SamplerElement); RPS_sampler_elem_reinit (s); - LOG (GNUNET_ERROR_TYPE_DEBUG, "SAMPLER: initialised with empty PeerID\n"); return s; } /** - * Input an PeerID into the given sampler. + * Input an PeerID into the given sampler element. + * + * @param sampler the sampler the @a s_elem belongs to. + * Needed to know the */ - static void -RPS_sampler_elem_next (struct RPS_SamplerElement *s_elem, const struct GNUNET_PeerIdentity *other, - RPS_sampler_insert_cb insert_cb, void *insert_cls, - RPS_sampler_remove_cb remove_cb, void *remove_cls) +static void +RPS_sampler_elem_next (struct RPS_SamplerElement *s_elem, + struct RPS_Sampler *sampler, /* TODO remove? */ + const struct GNUNET_PeerIdentity *other) { struct GNUNET_HashCode other_hash; - if ( 0 == GNUNET_CRYPTO_cmp_peer_identity(other, &(s_elem->peer_id)) ) + s_elem->num_peers++; + + to_file (s_elem->file_name, + "Got id %s", + GNUNET_i2s_full (other)); + + if (0 == GNUNET_CRYPTO_cmp_peer_identity (other, &(s_elem->peer_id))) { - LOG(GNUNET_ERROR_TYPE_DEBUG, "SAMPLER: Got PeerID %s_elem\n", - GNUNET_i2s(other)); - LOG(GNUNET_ERROR_TYPE_DEBUG, "SAMPLER: Have already PeerID %s_elem\n", - GNUNET_i2s(&(s_elem->peer_id))); + LOG (GNUNET_ERROR_TYPE_DEBUG, " Got PeerID %s\n", + GNUNET_i2s (other)); + LOG (GNUNET_ERROR_TYPE_DEBUG, "Have already PeerID %s\n", + GNUNET_i2s (&(s_elem->peer_id))); } else { @@ -283,130 +438,168 @@ RPS_sampler_elem_next (struct RPS_SamplerElement *s_elem, const struct GNUNET_Pe sizeof(struct GNUNET_PeerIdentity), &other_hash); - if ( EMPTY == s_elem->is_empty ) - { // Or whatever is a valid way to say - // "we have no PeerID at the moment" - LOG(GNUNET_ERROR_TYPE_DEBUG, "SAMPLER: Got PeerID %s_elem; Simply accepting (was empty previously).\n", - GNUNET_i2s(other)); + if (EMPTY == s_elem->is_empty) + { + LOG (GNUNET_ERROR_TYPE_DEBUG, + "Got PeerID %s; Simply accepting (was empty previously).\n", + GNUNET_i2s(other)); s_elem->peer_id = *other; - //s_elem->peer_id = other; s_elem->peer_id_hash = other_hash; - if (NULL != sampler->insert_cb) - { - sampler->insert_cb(sampler->insert_cls, &(s_elem->peer_id)); - } + + s_elem->num_change++; } - else if ( 0 > GNUNET_CRYPTO_hash_cmp(&other_hash, &s_elem->peer_id_hash) ) + else if (0 > GNUNET_CRYPTO_hash_cmp (&other_hash, &s_elem->peer_id_hash)) { - LOG(GNUNET_ERROR_TYPE_DEBUG, "SAMPLER: Got PeerID %s_elem\n", - GNUNET_i2s(other)); - LOG(GNUNET_ERROR_TYPE_DEBUG, "SAMPLER: Discarding old PeerID %s_elem\n", - GNUNET_i2s(&s_elem->peer_id)); - - if ( NULL != sampler->remove_cb ) - { - LOG(GNUNET_ERROR_TYPE_DEBUG, "SAMPLER: Removing old PeerID %s_elem with the remove callback.\n", - GNUNET_i2s(&s_elem->peer_id)); - sampler->remove_cb(sampler->remove_cls, &s_elem->peer_id); - } - - memcpy(&s_elem->peer_id, other, sizeof(struct GNUNET_PeerIdentity)); - //s_elem->peer_id = other; + LOG (GNUNET_ERROR_TYPE_DEBUG, " Got PeerID %s\n", + GNUNET_i2s (other)); + LOG (GNUNET_ERROR_TYPE_DEBUG, "Discarding old PeerID %s\n", + GNUNET_i2s (&s_elem->peer_id)); + s_elem->peer_id = *other; s_elem->peer_id_hash = other_hash; - if ( NULL != sampler->insert_cb ) - { - LOG(GNUNET_ERROR_TYPE_DEBUG, "SAMPLER: Inserting new PeerID %s_elem with the insert callback.\n", - GNUNET_i2s(&s_elem->peer_id)); - sampler->insert_cb(sampler->insert_cls, &s_elem->peer_id); - } + s_elem->num_change++; } else { - LOG(GNUNET_ERROR_TYPE_DEBUG, "SAMPLER: Got PeerID %s_elem\n", - GNUNET_i2s(other)); - LOG(GNUNET_ERROR_TYPE_DEBUG, "SAMPLER: Keeping old PeerID %s_elem\n", - GNUNET_i2s(&s_elem->peer_id)); + LOG (GNUNET_ERROR_TYPE_DEBUG, " Got PeerID %s\n", + GNUNET_i2s (other)); + LOG (GNUNET_ERROR_TYPE_DEBUG, "Keeping old PeerID %s\n", + GNUNET_i2s (&s_elem->peer_id)); } } s_elem->is_empty = NOT_EMPTY; + + to_file (s_elem->file_name, + "Now holding %s", + GNUNET_i2s_full (&s_elem->peer_id)); +} + + +/** + * Get the size of the sampler. + * + * @param sampler the sampler to return the size of. + * @return the size of the sampler + */ +unsigned int +RPS_sampler_get_size (struct RPS_Sampler *sampler) +{ + return sampler->sampler_size; } /** * Grow or shrink the size of the sampler. * + * @param sampler the sampler to resize. * @param new_size the new size of the sampler */ - void -RPS_sampler_resize (unsigned int new_size) +static void +sampler_resize (struct RPS_Sampler *sampler, unsigned int new_size) { unsigned int old_size; - uint64_t i; - struct RPS_SamplerElement **rem_list; + uint32_t i; // TODO check min and max size old_size = sampler->sampler_size; - if (old_size > new_size*4 && - extra_size > new_size*4) + if (old_size > new_size) { /* Shrinking */ - new_size /= 2; - - /* Temporary store those to properly call the removeCB on those later */ - rem_list = GNUNET_malloc ((old_size - new_size) * sizeof (struct RPS_SamplerElement *)); - memcpy (rem_list, - &sampler->sampler_elements[new_size], - (old_size - new_size) * sizeof(struct RPS_SamplerElement *)); - - LOG (GNUNET_ERROR_TYPE_DEBUG, "SAMPLER: Shrinking sampler %d -> %d\n", old_size, new_size); - GNUNET_array_grow (sampler->sampler_elements, sampler->sampler_size, new_size); LOG (GNUNET_ERROR_TYPE_DEBUG, - "SAMPLER: sampler->sampler_elements now points to %p\n", - sampler->sampler_elements); - - // TODO move extended_samplers_index + "Shrinking sampler %d -> %d\n", + old_size, + new_size); + + to_file (sampler->file_name, + "Shrinking sampler %d -> %d", + old_size, + new_size); + + /* TODO Temporary store those to properly call the removeCB on those later? */ + GNUNET_array_grow (sampler->trash_can, + sampler->trash_can_size, + old_size - new_size); for (i = new_size ; i < old_size ; i++) - {/* Remove unneeded rest */ - LOG (GNUNET_ERROR_TYPE_DEBUG, "SAMPLER: Removing %" PRIX64 ". sampler\n", i); - if (NULL != sampler->remove_cb) - sampler->remove_cb (sampler->remove_cls, &rem_list[i]->peer_id); - GNUNET_free (rem_list[i]); + { + to_file (sampler->file_name, + "-%" PRIu32 ": %s", + i, + sampler->sampler_elements[i]->file_name); + to_file (sampler->sampler_elements[i]->file_name, + "--- non-active"); + sampler->trash_can[i - new_size] = sampler->sampler_elements[i]; } + + GNUNET_array_grow (sampler->sampler_elements, + sampler->sampler_size, + new_size); + LOG (GNUNET_ERROR_TYPE_DEBUG, + "sampler->sampler_elements now points to %p\n", + sampler->sampler_elements); + } - else if (old_size < new_size)// || - //extra_size < new_size) // needed? + else if (old_size < new_size) { /* Growing */ - new_size *= 2; // TODO check overflow - - LOG (GNUNET_ERROR_TYPE_DEBUG, "SAMPLER: Growing sampler %d -> %d\n", old_size, new_size); - GNUNET_array_grow (sampler->sampler_elements, sampler->sampler_size, new_size); LOG (GNUNET_ERROR_TYPE_DEBUG, - "SAMPLER: sampler->sampler_elements now points to %p\n", - sampler->sampler_elements); + "Growing sampler %d -> %d\n", + old_size, + new_size); + + to_file (sampler->file_name, + "Growing sampler %d -> %d", + old_size, + new_size); - // TODO move extended_samplers_index + GNUNET_array_grow (sampler->sampler_elements, + sampler->sampler_size, + new_size); - for ( i = old_size ; i < new_size ; i++ ) + for (i = old_size ; i < new_size ; i++) { /* Add new sampler elements */ sampler->sampler_elements[i] = RPS_sampler_elem_create (); - if (NULL != sampler->insert_cb) - sampler->insert_cb (sampler->insert_cls, &sampler->sampler_elements[i]->peer_id); - LOG (GNUNET_ERROR_TYPE_DEBUG, - "SAMPLER: Added %" PRIX64 ". sampler, now pointing to %p, contains %s\n", - i, &sampler->sampler_elements[i], GNUNET_i2s (&sampler->sampler_elements[i]->peer_id)); + + to_file (sampler->file_name, + "+%" PRIu32 ": %s", + i, + sampler->sampler_elements[i]->file_name); } } else { - LOG (GNUNET_ERROR_TYPE_DEBUG, "SAMPLER: Size remains the same -- nothing to do\n"); + LOG (GNUNET_ERROR_TYPE_DEBUG, "Size remains the same -- nothing to do\n"); return; } - GNUNET_assert(sampler->sampler_size == new_size); - LOG(GNUNET_ERROR_TYPE_DEBUG, "SAMPLER: Finished growing/shrinking.\n"); // remove + GNUNET_assert (sampler->sampler_size == new_size); +} + + +/** + * Grow or shrink the size of the sampler. + * + * @param sampler the sampler to resize. + * @param new_size the new size of the sampler + */ +void +RPS_sampler_resize (struct RPS_Sampler *sampler, unsigned int new_size) +{ + GNUNET_assert (0 < new_size); + sampler_resize (sampler, new_size); +} + + +/** + * Empty the sampler. + * + * @param sampler the sampler to empty. + * @param new_size the new size of the sampler + */ +static void +sampler_empty (struct RPS_Sampler *sampler) +{ + sampler_resize (sampler, 0); } @@ -414,59 +607,102 @@ RPS_sampler_resize (unsigned int new_size) * Initialise a tuple of sampler elements. * * @param init_size the size the sampler is initialised with - * @param id with which all newly created sampler elements are initialised - * @param ins_cb the callback that will be called on every PeerID that is - * newly inserted into a sampler element - * @param ins_cls the closure given to #ins_cb - * @param rem_cb the callback that will be called on every PeerID that is - * removed from a sampler element - * @param rem_cls the closure given to #rem_cb + * @param max_round_interval maximum time a round takes + * @return a handle to a sampler that consists of sampler elements. */ - void -RPS_sampler_init (size_t init_size, const struct GNUNET_PeerIdentity *id, - RPS_sampler_insert_cb ins_cb, void *ins_cls, - RPS_sampler_remove_cb rem_cb, void *rem_cls) +struct RPS_Sampler * +RPS_sampler_init (size_t init_size, + struct GNUNET_TIME_Relative max_round_interval) { - //struct RPS_Sampler *sampler; - //uint64_t i; + struct RPS_Sampler *sampler; /* Initialise context around extended sampler */ min_size = 10; // TODO make input to _samplers_init() max_size = 1000; // TODO make input to _samplers_init() - GNUNET_new_array (64, struct GNUNET_TIME_Relative); sampler = GNUNET_new (struct RPS_Sampler); + + #ifdef TO_FILE + sampler->file_name = create_file ("sampler-"); + + LOG (GNUNET_ERROR_TYPE_DEBUG, + "Initialised sampler %s\n", + sampler->file_name); + #endif /* TO_FILE */ + sampler->sampler_size = 0; sampler->sampler_elements = NULL; - sampler->insert_cb = ins_cb; - sampler->insert_cls = ins_cls; - sampler->remove_cb = rem_cb; - sampler->remove_cls = rem_cls; + sampler->trash_can_size = 0; + sampler->trash_can = NULL; + sampler->max_round_interval = max_round_interval; + sampler->get_peers = sampler_get_rand_peer; + sampler->gpc_head = NULL; + sampler->gpc_tail = NULL; //sampler->sampler_elements = GNUNET_new_array(init_size, struct GNUNET_PeerIdentity); //GNUNET_array_grow (sampler->sampler_elements, sampler->sampler_size, min_size); - RPS_sampler_resize (init_size); - RPS_sampler_update_list (id); // no super nice desing but ok for the moment + RPS_sampler_resize (sampler, init_size); - extended_samplers_index = sampler->sampler_elements; + client_get_index = 0; //GNUNET_assert (init_size == sampler->sampler_size); + return sampler; +} + +/** + * Initialise a modified tuple of sampler elements. + * + * @param init_size the size the sampler is initialised with + * @param max_round_interval maximum time a round takes + * @return a handle to a sampler that consists of sampler elements. + */ +struct RPS_Sampler * +RPS_sampler_mod_init (size_t init_size, + struct GNUNET_TIME_Relative max_round_interval) +{ + struct RPS_Sampler *sampler; + + sampler = RPS_sampler_init (init_size, max_round_interval); + sampler->get_peers = sampler_mod_get_rand_peer; + + LOG (GNUNET_ERROR_TYPE_DEBUG, + "Initialised modified sampler %s\n", + sampler->file_name); + to_file (sampler->file_name, + "This is a modified sampler"); + + return sampler; } /** * A fuction to update every sampler in the given list * + * @param sampler the sampler to update. * @param id the PeerID that is put in the sampler */ void -RPS_sampler_update_list (const struct GNUNET_PeerIdentity *id) +RPS_sampler_update (struct RPS_Sampler *sampler, + const struct GNUNET_PeerIdentity *id) { - uint64_t i; + uint32_t i; - for ( i = 0 ; i < sampler->sampler_size ; i++ ) - RPS_sampler_elem_next (sampler->sampler_elements[i], id, - sampler->insert_cb, sampler->insert_cls, - sampler->remove_cb, sampler->remove_cls); + to_file (sampler->file_name, + "Got %s", + GNUNET_i2s_full (id)); + + for (i = 0 ; i < sampler->sampler_size ; i++) + { + RPS_sampler_elem_next (sampler->sampler_elements[i], + sampler, + id); + } + + for (i = 0 ; i < sampler->trash_can_size ; i++) + { + RPS_sampler_elem_next (sampler->trash_can[i], + sampler, + id); + } } @@ -475,18 +711,28 @@ RPS_sampler_update_list (const struct GNUNET_PeerIdentity *id) * * Used to get rid of a PeerID. * + * @param sampler the sampler to reinitialise a sampler element in. * @param id the id of the sampler elements to update. */ void -RPS_sampler_reinitialise_by_value (const struct GNUNET_PeerIdentity *id) +RPS_sampler_reinitialise_by_value (struct RPS_Sampler *sampler, + const struct GNUNET_PeerIdentity *id) { - uint64_t i; + uint32_t i; + struct RPS_SamplerElement *trash_entry; for ( i = 0 ; i < sampler->sampler_size ; i++ ) { if ( 0 == GNUNET_CRYPTO_cmp_peer_identity(id, &(sampler->sampler_elements[i]->peer_id)) ) { - LOG(GNUNET_ERROR_TYPE_DEBUG, "SAMPLER: Reinitialising sampler\n"); + LOG (GNUNET_ERROR_TYPE_DEBUG, "Reinitialising sampler\n"); + trash_entry = GNUNET_new (struct RPS_SamplerElement); + *trash_entry = *(sampler->sampler_elements[i]); + GNUNET_array_append (sampler->trash_can, + sampler->trash_can_size, + trash_entry); + to_file (trash_entry->file_name, + "--- non-active"); RPS_sampler_elem_reinit (sampler->sampler_elements[i]); } } @@ -500,105 +746,157 @@ RPS_sampler_reinitialise_by_value (const struct GNUNET_PeerIdentity *id) * corrsponding peer to the client. * Only used internally */ - const struct GNUNET_PeerIdentity * -RPS_sampler_get_rand_peer_ () +static void +sampler_get_rand_peer (void *cls, + const struct GNUNET_SCHEDULER_TaskContext *tc) { - uint64_t r_index; - const struct GNUNET_PeerIdentity *peer; // do we have to malloc that? + struct GetPeerCls *gpc = cls; + uint32_t r_index; - // TODO implement extra logic + gpc->get_peer_task = NULL; + if (0 != (tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN)) + return; /**; * Choose the r_index of the peer we want to return * at random from the interval of the gossip list */ - r_index = GNUNET_CRYPTO_random_u64(GNUNET_CRYPTO_QUALITY_STRONG, - sampler->sampler_size); + r_index = GNUNET_CRYPTO_random_u64 (GNUNET_CRYPTO_QUALITY_STRONG, + gpc->sampler->sampler_size); - //if ( EMPTY == sampler->sampler_elements[r_index]->is_empty ) - // // TODO schedule for later - // peer = NULL; - //else - peer = &(sampler->sampler_elements[r_index]->peer_id); - sampler->sampler_elements[r_index]->last_request = GNUNET_TIME_absolute_get(); - LOG(GNUNET_ERROR_TYPE_DEBUG, "Sgrp: Returning PeerID %s\n", GNUNET_i2s(peer)); + if (EMPTY == gpc->sampler->sampler_elements[r_index]->is_empty) + { + //LOG (GNUNET_ERROR_TYPE_DEBUG, + // "Not returning randomly selected, empty PeerID. - Rescheduling.\n"); + + /* FIXME no active wait - get notified, when new id arrives? + * Might also be a freshly emptied one. Others might still contain ids. + * Counter? + */ + gpc->get_peer_task = + GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_relative_multiply ( + GNUNET_TIME_UNIT_SECONDS, 0.1), + &sampler_get_rand_peer, + cls); + return; + } + + *gpc->id = gpc->sampler->sampler_elements[r_index]->peer_id; + gpc->cont (gpc->cont_cls, gpc->id); - return peer; + GNUNET_CONTAINER_DLL_remove (gpc->sampler->gpc_head, + gpc->sampler->gpc_tail, + gpc); + + GNUNET_free (gpc); } + /** - * Get n random peers out of the sampled peers. + * Get one random peer out of the sampled peers. * * We might want to reinitialise this sampler after giving the * corrsponding peer to the client. - * Random with or without consumption? - * Only used internally */ - const struct GNUNET_PeerIdentity * -RPS_sampler_get_n_rand_peers_ (uint64_t n) +static void +sampler_mod_get_rand_peer (void *cls, + const struct GNUNET_SCHEDULER_TaskContext *tc) { - if ( 0 == sampler->sampler_size ) - { - LOG (GNUNET_ERROR_TYPE_DEBUG, - "Sgrp: List empty - Returning NULL\n"); - return NULL; - } - else - { - // TODO check if we have too much (distinct) sampled peers - // If we are not ready yet maybe schedule for later - struct GNUNET_PeerIdentity *peers; - uint64_t i; + struct GetPeerCls *gpc = cls; + struct GNUNET_PeerIdentity tmp_id; + unsigned int empty_flag; + struct RPS_SamplerElement *s_elem; + struct GNUNET_TIME_Relative last_request_diff; + uint32_t tmp_client_get_index; + + gpc->get_peer_task = NULL; + if (0 != (tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN)) + return; - peers = GNUNET_malloc (n * sizeof(struct GNUNET_PeerIdentity)); + LOG (GNUNET_ERROR_TYPE_DEBUG, "Single peer was requested\n"); - for ( i = 0 ; i < n ; i++ ) { - //peers[i] = RPS_sampler_get_rand_peer_(sampler->sampler_elements); - memcpy (&peers[i], RPS_sampler_get_rand_peer_ (), sizeof (struct GNUNET_PeerIdentity)); + + /* Store the next #client_get_index to check whether we cycled over the whole list */ + if (0 < client_get_index) + tmp_client_get_index = client_get_index - 1; + else + tmp_client_get_index = gpc->sampler->sampler_size - 1; + + LOG (GNUNET_ERROR_TYPE_DEBUG, + "sched for later if index reaches %" PRIX32 " (sampler size: %" PRIX32 ").\n", + tmp_client_get_index, gpc->sampler->sampler_size); + + do + { /* Get first non empty sampler */ + if (tmp_client_get_index == client_get_index) + { /* We once cycled over the whole list */ + LOG (GNUNET_ERROR_TYPE_DEBUG, "reached tmp_index %" PRIX32 ".\n", + client_get_index); + GNUNET_assert (NULL == gpc->get_peer_task); + gpc->get_peer_task = + GNUNET_SCHEDULER_add_delayed (gpc->sampler->max_round_interval, + &sampler_mod_get_rand_peer, + cls); + return; } - return peers; - } -} + tmp_id = gpc->sampler->sampler_elements[client_get_index]->peer_id; + empty_flag = gpc->sampler->sampler_elements[client_get_index]->is_empty; + RPS_sampler_elem_reinit (gpc->sampler->sampler_elements[client_get_index]); + if (EMPTY != empty_flag) + RPS_sampler_elem_next (gpc->sampler->sampler_elements[client_get_index], + gpc->sampler, + &tmp_id); + + /* Cycle the #client_get_index one step further */ + if ( client_get_index == gpc->sampler->sampler_size - 1 ) + client_get_index = 0; + else + client_get_index++; -/** - * Get one random peer out of the sampled peers. - * - * We might want to reinitialise this sampler after giving the - * corrsponding peer to the client. - * - * @return a random PeerID of the PeerIDs previously put into the sampler. - */ - const struct GNUNET_PeerIdentity * -RPS_sampler_get_rand_peer () -{ - struct GNUNET_PeerIdentity *peer; + /* LOG (GNUNET_ERROR_TYPE_DEBUG, "incremented index to %" PRIX32 ".\n", + client_get_index); */ + } while (EMPTY == gpc->sampler->sampler_elements[client_get_index]->is_empty); + + s_elem = gpc->sampler->sampler_elements[client_get_index]; + *gpc->id = s_elem->peer_id; - if (64 > req_counter) - req_counter++; - if (1 < req_counter) + /* Check whether we may use this sampler to give it back to the client */ + if (GNUNET_TIME_UNIT_FOREVER_ABS.abs_value_us != s_elem->last_client_request.abs_value_us) { - memcpy (&request_deltas[1], - request_deltas, - (req_counter - 1) * sizeof (struct GNUNET_TIME_Relative)); - request_deltas[0] = GNUNET_TIME_absolute_get_difference (last_request, - GNUNET_TIME_absolute_get ()); - request_rate = T_relative_avg (request_deltas, req_counter); + last_request_diff = + GNUNET_TIME_absolute_get_difference (s_elem->last_client_request, + GNUNET_TIME_absolute_get ()); + /* We're not going to give it back now if it was + * already requested by a client this round */ + if (last_request_diff.rel_value_us < gpc->sampler->max_round_interval.rel_value_us) + { + LOG (GNUNET_ERROR_TYPE_DEBUG, + "Last client request on this sampler was less than max round interval ago -- scheduling for later\n"); + ///* How many time remains untile the next round has started? */ + //inv_last_request_diff = + // GNUNET_TIME_absolute_get_difference (last_request_diff, + // sampler->max_round_interval); + // add a little delay + /* Schedule it one round later */ + GNUNET_assert (NULL == gpc->get_peer_task); + gpc->get_peer_task = + GNUNET_SCHEDULER_add_delayed (gpc->sampler->max_round_interval, + &sampler_mod_get_rand_peer, + cls); + return; + } + // TODO add other reasons to wait here } - last_request = GNUNET_TIME_absolute_get(); - // TODO resize the size of the extended_samplers - - // use _get_rand_peer_ ? - peer = GNUNET_new (struct GNUNET_PeerIdentity); - *peer = (*extended_samplers_index)->peer_id; - RPS_sampler_elem_reinit (*extended_samplers_index); - if ( extended_samplers_index == &sampler->sampler_elements[sampler->sampler_size -1] ) - extended_samplers_index = &sampler->sampler_elements[0]; - else - extended_samplers_index++; - // TODO - return peer; + + s_elem->last_client_request = GNUNET_TIME_absolute_get (); + + GNUNET_CONTAINER_DLL_remove (gpc->sampler->gpc_head, + gpc->sampler->gpc_tail, + gpc); + gpc->cont (gpc->cont_cls, gpc->id); + GNUNET_free (gpc); } @@ -609,32 +907,51 @@ RPS_sampler_get_rand_peer () * corrsponding peer to the client. * Random with or without consumption? * - * @return n random PeerIDs of the PeerIDs previously put into the sampler. + * @param sampler the sampler to get peers from. + * @param cb callback that will be called once the ids are ready. + * @param cls closure given to @a cb + * @param for_client #GNUNET_YES if result is used for client, + * #GNUNET_NO if used internally + * @param num_peers the number of peers requested */ - const struct GNUNET_PeerIdentity * -RPS_sampler_get_n_rand_peers (uint64_t n) + void +RPS_sampler_get_n_rand_peers (struct RPS_Sampler *sampler, + RPS_sampler_n_rand_peers_ready_cb cb, + void *cls, uint32_t num_peers) { - // use _get_rand_peers_ ? - if ( 0 == sampler->sampler_size ) - { - LOG (GNUNET_ERROR_TYPE_DEBUG, - "Sgrp: List empty - Returning NULL\n"); - return NULL; - } - else - { - // TODO check if we have too much (distinct) sampled peers - // If we are not ready yet maybe schedule for later - struct GNUNET_PeerIdentity *peers; - uint64_t i; + GNUNET_assert (0 != sampler->sampler_size); + if (0 == num_peers) + return; - peers = GNUNET_malloc (n * sizeof(struct GNUNET_PeerIdentity)); + // TODO check if we have too much (distinct) sampled peers + uint32_t i; + struct NRandPeersReadyCls *cb_cls; + struct GetPeerCls *gpc; - for ( i = 0 ; i < n ; i++ ) { - //peers[i] = RPS_sampler_get_rand_peer_(sampler->sampler_elements); - memcpy (&peers[i], RPS_sampler_get_rand_peer (), sizeof (struct GNUNET_PeerIdentity)); - } - return peers; + cb_cls = GNUNET_new (struct NRandPeersReadyCls); + cb_cls->num_peers = num_peers; + cb_cls->cur_num_peers = 0; + cb_cls->ids = GNUNET_new_array (num_peers, struct GNUNET_PeerIdentity); + cb_cls->callback = cb; + cb_cls->cls = cls; + + LOG (GNUNET_ERROR_TYPE_DEBUG, + "Scheduling requests for %" PRIu32 " peers\n", num_peers); + + for (i = 0 ; i < num_peers ; i++) + { + gpc = GNUNET_new (struct GetPeerCls); + gpc->sampler = sampler; + gpc->cont = check_n_peers_ready; + gpc->cont_cls = cb_cls; + gpc->id = &cb_cls->ids[i]; + + // maybe add a little delay + gpc->get_peer_task = GNUNET_SCHEDULER_add_now (sampler->get_peers, gpc); + + GNUNET_CONTAINER_DLL_insert (sampler->gpc_head, + sampler->gpc_tail, + gpc); } } @@ -642,20 +959,22 @@ RPS_sampler_get_n_rand_peers (uint64_t n) /** * Counts how many Samplers currently hold a given PeerID. * + * @param sampler the sampler to count ids in. * @param id the PeerID to count. * * @return the number of occurrences of id. */ - uint64_t -RPS_sampler_count_id (const struct GNUNET_PeerIdentity *id) + uint32_t +RPS_sampler_count_id (struct RPS_Sampler *sampler, + const struct GNUNET_PeerIdentity *id) { - uint64_t count; - uint64_t i; + uint32_t count; + uint32_t i; count = 0; for ( i = 0 ; i < sampler->sampler_size ; i++ ) { - if ( 0 == GNUNET_CRYPTO_cmp_peer_identity (&sampler->sampler_elements[i]->peer_id, id) + if ( 0 == GNUNET_CRYPTO_cmp_peer_identity (&sampler->sampler_elements[i]->peer_id, id) && EMPTY != sampler->sampler_elements[i]->is_empty) count++; } @@ -667,11 +986,21 @@ RPS_sampler_count_id (const struct GNUNET_PeerIdentity *id) * Cleans the sampler. */ void -RPS_sampler_destroy () +RPS_sampler_destroy (struct RPS_Sampler *sampler) { - RPS_sampler_resize (0); - GNUNET_free (request_deltas); // _array_grow()? - GNUNET_array_grow (sampler->sampler_elements, sampler->sampler_size, 0); + struct GetPeerCls *i; + + for (i = sampler->gpc_head; NULL != i; i = sampler->gpc_head) + { + GNUNET_CONTAINER_DLL_remove (sampler->gpc_head, + sampler->gpc_tail, + i); + GNUNET_SCHEDULER_cancel (i->get_peer_task); + GNUNET_free (i); + } + + sampler_empty (sampler); + GNUNET_free (sampler); } /* end of gnunet-service-rps.c */