X-Git-Url: https://git.librecmc.org/?a=blobdiff_plain;f=src%2Frps%2Fgnunet-service-rps_sampler.c;h=b51867d28b942e32b8edd392e08ec419451ef767;hb=36fd53ed6ae7c60d4b49cea614d18aa6d58843a1;hp=a0624c9421bb165de9a7dbf49aedfed7d332ed76;hpb=2f61fb7ffc7232c69e66a13c956892d6f6d13081;p=oweals%2Fgnunet.git diff --git a/src/rps/gnunet-service-rps_sampler.c b/src/rps/gnunet-service-rps_sampler.c index a0624c942..b51867d28 100644 --- a/src/rps/gnunet-service-rps_sampler.c +++ b/src/rps/gnunet-service-rps_sampler.c @@ -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,8 +32,11 @@ #include #include +#include "rps-test_util.h" + #define LOG(kind, ...) GNUNET_log_from(kind,"rps-sampler",__VA_ARGS__) + // multiple 'clients'? // TODO check for overflows @@ -111,6 +114,59 @@ struct RPS_SamplerElement #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. @@ -127,7 +183,7 @@ typedef void * Only used internally */ static void -sampler_get_rand_peer2 (void *cls, +sampler_get_rand_peer (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc); /** @@ -137,7 +193,7 @@ sampler_get_rand_peer2 (void *cls, * corrsponding peer to the client. */ static void -sampler_get_rand_peer (void *cls, +sampler_mod_get_rand_peer (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc); @@ -153,10 +209,29 @@ struct RPS_Sampler //size_t size; /** - * All Samplers in one array. + * All sampler elements in one array. */ struct RPS_SamplerElement **sampler_elements; + /** + * 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 * @@ -170,6 +245,16 @@ struct RPS_Sampler */ RPS_get_peers_type get_peers; + /** + * Head for the DLL to store the closures to pending requests. + */ + struct GetPeerCls *gpc_head; + + /** + * Tail for the DLL to store the closures to pending requests. + */ + struct GetPeerCls *gpc_tail; + #ifdef TO_FILE /** * File name to log to @@ -209,58 +294,6 @@ struct NRandPeersReadyCls void *cls; }; -/** - * 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_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; -}; - - ///** // * Global sampler variable. // */ @@ -288,76 +321,6 @@ static size_t max_size; static uint32_t client_get_index; -#ifdef TO_FILE -/** - * This function is used to facilitate writing important information to disk - */ -#define to_file(file_name, ...) do {char tmp_buf[512];\ - int size;\ - size = GNUNET_snprintf(tmp_buf,sizeof(tmp_buf),__VA_ARGS__);\ - if (0 > size)\ - LOG (GNUNET_ERROR_TYPE_WARNING,\ - "Failed to create tmp_buf\n");\ - else\ - to_file_(file_name,tmp_buf);\ -} while (0); - -static void -to_file_ (char *file_name, char *line) -{ - struct GNUNET_DISK_FileHandle *f; - char output_buffer[512]; - //size_t size; - int size; - size_t size2; - - - if (NULL == (f = GNUNET_DISK_file_open (file_name, - GNUNET_DISK_OPEN_APPEND | - GNUNET_DISK_OPEN_WRITE | - GNUNET_DISK_OPEN_CREATE, - GNUNET_DISK_PERM_USER_WRITE))) - { - LOG (GNUNET_ERROR_TYPE_WARNING, - "Not able to open file %s\n", - file_name); - return; - } - size = GNUNET_snprintf (output_buffer, - sizeof (output_buffer), - "%llu %s\n", - GNUNET_TIME_absolute_get ().abs_value_us, - line); - if (0 > size) - { - LOG (GNUNET_ERROR_TYPE_WARNING, - "Failed to write string to buffer (size: %i)\n", - size); - return; - } - - size2 = GNUNET_DISK_file_write (f, output_buffer, size); - if (size != size2) - { - LOG (GNUNET_ERROR_TYPE_WARNING, - "Unable to write to file! (Size: %u, size2: %u)\n", - size, - size2); - return; - } - - if (GNUNET_YES != GNUNET_DISK_file_close (f)) - LOG (GNUNET_ERROR_TYPE_WARNING, - "Unable to close file\n"); -} -#endif /* TO_FILE */ - - -/** FIXME document */ -static struct GetPeerCls *gpc_head; -static struct GetPeerCls *gpc_tail; - - /** * Callback to _get_rand_peer() used by _get_n_rand_peers(). * @@ -406,38 +369,11 @@ RPS_sampler_elem_reinit (struct RPS_SamplerElement *sampler_el) #ifdef TO_FILE /* Create a file(-name) to store internals to */ - int size; - char *end; - char *buf; - char name_buf[512]; - size_t keylen = (sizeof (struct GNUNET_CRYPTO_AuthKey)) * 8; - - if (keylen % 5 > 0) - keylen += 5 - keylen % 5; - keylen /= 5; - buf = GNUNET_malloc (keylen + 1); - - end = GNUNET_STRINGS_data_to_string (&(sampler_el->auth_key.key), - sizeof (struct GNUNET_CRYPTO_AuthKey), - buf, - keylen); - - if (NULL == end) - { - GNUNET_free (buf); - GNUNET_break (0); - } - else - { - *end = '\0'; - } - - size = GNUNET_snprintf (name_buf, sizeof (name_buf), "sampler_el-%s-", buf); - if (0 > size) - LOG (GNUNET_ERROR_TYPE_WARNING, "Failed to create name_buf\n"); + char *name_buf; + name_buf = auth_key_to_string (sampler_el->auth_key); - if (NULL == (sampler_el->file_name = GNUNET_DISK_mktemp (name_buf))) - LOG (GNUNET_ERROR_TYPE_WARNING, "Could not create file\n"); + sampler_el->file_name = create_file (name_buf); + GNUNET_free (name_buf); #endif /* TO_FILE */ sampler_el->last_client_request = GNUNET_TIME_UNIT_FOREVER_ABS; @@ -477,18 +413,16 @@ RPS_sampler_elem_create (void) */ static void RPS_sampler_elem_next (struct RPS_SamplerElement *s_elem, - struct RPS_Sampler *sampler, + struct RPS_Sampler *sampler, /* TODO remove? */ const struct GNUNET_PeerIdentity *other) { struct GNUNET_HashCode other_hash; s_elem->num_peers++; - #ifdef TO_FILE to_file (s_elem->file_name, "Got id %s", GNUNET_i2s_full (other)); - #endif /* TO_FILE */ if (0 == GNUNET_CRYPTO_cmp_peer_identity (other, &(s_elem->peer_id))) { @@ -534,11 +468,10 @@ RPS_sampler_elem_next (struct RPS_SamplerElement *s_elem, } } s_elem->is_empty = NOT_EMPTY; - #ifdef TO_FILE + to_file (s_elem->file_name, "Now holding %s", GNUNET_i2s_full (&s_elem->peer_id)); - #endif /* TO_FILE */ } @@ -573,32 +506,38 @@ sampler_resize (struct RPS_Sampler *sampler, unsigned int new_size) if (old_size > new_size) { /* Shrinking */ - /* Temporary store those to properly call the removeCB on those later */ LOG (GNUNET_ERROR_TYPE_DEBUG, "Shrinking sampler %d -> %d\n", old_size, new_size); - #ifdef TO_FILE + 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++) { 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]; } - #endif /* TO_FILE */ + GNUNET_array_grow (sampler->sampler_elements, - sampler->sampler_size, - new_size); + sampler->sampler_size, + new_size); LOG (GNUNET_ERROR_TYPE_DEBUG, - "sampler->sampler_elements now points to %p\n", - sampler->sampler_elements); + "sampler->sampler_elements now points to %p\n", + sampler->sampler_elements); } else if (old_size < new_size) @@ -607,12 +546,12 @@ sampler_resize (struct RPS_Sampler *sampler, unsigned int new_size) "Growing sampler %d -> %d\n", old_size, new_size); - #ifdef TO_FILE + to_file (sampler->file_name, "Growing sampler %d -> %d", old_size, new_size); - #endif /* TO_FILE */ + GNUNET_array_grow (sampler->sampler_elements, sampler->sampler_size, new_size); @@ -620,12 +559,11 @@ sampler_resize (struct RPS_Sampler *sampler, unsigned int new_size) for (i = old_size ; i < new_size ; i++) { /* Add new sampler elements */ sampler->sampler_elements[i] = RPS_sampler_elem_create (); - #ifdef TO_FILE + to_file (sampler->file_name, "+%" PRIu32 ": %s", i, sampler->sampler_elements[i]->file_name); - #endif /* TO_FILE */ } } else @@ -685,9 +623,8 @@ RPS_sampler_init (size_t init_size, sampler = GNUNET_new (struct RPS_Sampler); #ifdef TO_FILE - if (NULL == (sampler->file_name = GNUNET_DISK_mktemp ("sampler-"))) - LOG (GNUNET_ERROR_TYPE_WARNING, - "Could not create file\n"); + sampler->file_name = create_file ("sampler-"); + LOG (GNUNET_ERROR_TYPE_DEBUG, "Initialised sampler %s\n", sampler->file_name); @@ -695,8 +632,12 @@ RPS_sampler_init (size_t init_size, sampler->sampler_size = 0; sampler->sampler_elements = NULL; + 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 (sampler, init_size); @@ -721,15 +662,13 @@ RPS_sampler_mod_init (size_t init_size, struct RPS_Sampler *sampler; sampler = RPS_sampler_init (init_size, max_round_interval); - sampler->get_peers = sampler_get_rand_peer2; + sampler->get_peers = sampler_mod_get_rand_peer; - #ifdef TO_FILE LOG (GNUNET_ERROR_TYPE_DEBUG, "Initialised modified sampler %s\n", sampler->file_name); to_file (sampler->file_name, "This is a modified sampler"); - #endif /* TO_FILE */ return sampler; } @@ -747,11 +686,9 @@ RPS_sampler_update (struct RPS_Sampler *sampler, { uint32_t i; - #ifdef TO_FILE to_file (sampler->file_name, "Got %s", GNUNET_i2s_full (id)); - #endif /* TO_FILE */ for (i = 0 ; i < sampler->sampler_size ; i++) { @@ -759,6 +696,13 @@ RPS_sampler_update (struct RPS_Sampler *sampler, sampler, id); } + + for (i = 0 ; i < sampler->trash_can_size ; i++) + { + RPS_sampler_elem_next (sampler->trash_can[i], + sampler, + id); + } } @@ -775,12 +719,20 @@ RPS_sampler_reinitialise_by_value (struct RPS_Sampler *sampler, const struct GNUNET_PeerIdentity *id) { 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, "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]); } } @@ -795,14 +747,13 @@ RPS_sampler_reinitialise_by_value (struct RPS_Sampler *sampler, * Only used internally */ static void -sampler_get_rand_peer2 (void *cls, +sampler_get_rand_peer (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc) { struct GetPeerCls *gpc = cls; uint32_t r_index; gpc->get_peer_task = NULL; - GNUNET_CONTAINER_DLL_remove (gpc_head, gpc_tail, gpc); if (0 != (tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN)) return; @@ -813,19 +764,31 @@ sampler_get_rand_peer2 (void *cls, r_index = GNUNET_CRYPTO_random_u64 (GNUNET_CRYPTO_QUALITY_STRONG, gpc->sampler->sampler_size); - if ( EMPTY == gpc->sampler->sampler_elements[r_index]->is_empty ) + if (EMPTY == gpc->sampler->sampler_elements[r_index]->is_empty) { - gpc->get_peer_task = GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_relative_multiply( - GNUNET_TIME_UNIT_SECONDS, - .1), - &sampler_get_rand_peer2, - cls); + //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); + + GNUNET_CONTAINER_DLL_remove (gpc->sampler->gpc_head, + gpc->sampler->gpc_tail, + gpc); + GNUNET_free (gpc); } @@ -837,7 +800,7 @@ sampler_get_rand_peer2 (void *cls, * corrsponding peer to the client. */ static void -sampler_get_rand_peer (void *cls, +sampler_mod_get_rand_peer (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc) { struct GetPeerCls *gpc = cls; @@ -873,7 +836,7 @@ sampler_get_rand_peer (void *cls, GNUNET_assert (NULL == gpc->get_peer_task); gpc->get_peer_task = GNUNET_SCHEDULER_add_delayed (gpc->sampler->max_round_interval, - &sampler_get_rand_peer, + &sampler_mod_get_rand_peer, cls); return; } @@ -892,8 +855,8 @@ sampler_get_rand_peer (void *cls, else client_get_index++; - LOG (GNUNET_ERROR_TYPE_DEBUG, "incremented index to %" PRIX32 ".\n", - client_get_index); + /* 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]; @@ -920,7 +883,7 @@ sampler_get_rand_peer (void *cls, GNUNET_assert (NULL == gpc->get_peer_task); gpc->get_peer_task = GNUNET_SCHEDULER_add_delayed (gpc->sampler->max_round_interval, - &sampler_get_rand_peer, + &sampler_mod_get_rand_peer, cls); return; } @@ -929,7 +892,9 @@ sampler_get_rand_peer (void *cls, s_elem->last_client_request = GNUNET_TIME_absolute_get (); - GNUNET_CONTAINER_DLL_remove (gpc_head, gpc_tail, gpc); + GNUNET_CONTAINER_DLL_remove (gpc->sampler->gpc_head, + gpc->sampler->gpc_tail, + gpc); gpc->cont (gpc->cont_cls, gpc->id); GNUNET_free (gpc); } @@ -955,6 +920,8 @@ RPS_sampler_get_n_rand_peers (struct RPS_Sampler *sampler, void *cls, uint32_t num_peers) { GNUNET_assert (0 != sampler->sampler_size); + if (0 == num_peers) + return; // TODO check if we have too much (distinct) sampled peers uint32_t i; @@ -982,7 +949,9 @@ RPS_sampler_get_n_rand_peers (struct RPS_Sampler *sampler, // maybe add a little delay gpc->get_peer_task = GNUNET_SCHEDULER_add_now (sampler->get_peers, gpc); - GNUNET_CONTAINER_DLL_insert (gpc_head, gpc_tail, gpc); + GNUNET_CONTAINER_DLL_insert (sampler->gpc_head, + sampler->gpc_tail, + gpc); } } @@ -1021,9 +990,11 @@ RPS_sampler_destroy (struct RPS_Sampler *sampler) { struct GetPeerCls *i; - for (i = gpc_head; NULL != i; i = gpc_head) + for (i = sampler->gpc_head; NULL != i; i = sampler->gpc_head) { - GNUNET_CONTAINER_DLL_remove (gpc_head, gpc_tail, i); + GNUNET_CONTAINER_DLL_remove (sampler->gpc_head, + sampler->gpc_tail, + i); GNUNET_SCHEDULER_cancel (i->get_peer_task); GNUNET_free (i); }