2 This file is part of GNUnet.
5 GNUnet is free software: you can redistribute it and/or modify it
6 under the terms of the GNU General Public License as published
7 by the Free Software Foundation, either version 3 of the License,
8 or (at your option) any later version.
10 GNUnet is distributed in the hope that it will be useful, but
11 WITHOUT ANY WARRANTY; without even the implied warranty of
12 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
13 Affero General Public License for more details.
17 * @file rps/gnunet-service-rps_sampler.c
18 * @brief sampler implementation
19 * @author Julius Bünger
22 #include "gnunet_util_lib.h"
25 #include "gnunet-service-rps_sampler.h"
26 #include "gnunet-service-rps_sampler_elem.h"
31 #include "rps-test_util.h"
33 #define LOG(kind, ...) GNUNET_log_from(kind,"rps-sampler",__VA_ARGS__)
36 // multiple 'clients'?
38 // TODO check for overflows
40 // TODO align message structs
42 // hist_size_init, hist_size_max
44 /***********************************************************************
45 * WARNING: This section needs to be reviewed regarding the use of
46 * functions providing (pseudo)randomness!
47 ***********************************************************************/
49 // TODO care about invalid input of the caller (size 0 or less...)
52 * Callback that is called from _get_rand_peer() when the PeerID is ready.
54 * @param cls the closure given alongside this function.
55 * @param id the PeerID that was returned
58 (*RPS_sampler_rand_peer_ready_cont) (void *cls,
59 const struct GNUNET_PeerIdentity *id);
63 * Closure for #sampler_mod_get_rand_peer() and #sampler_get_rand_peer
70 struct GetPeerCls *next;
71 struct GetPeerCls *prev;
74 * The #RPS_SamplerRequestHandle this single request belongs to.
76 struct RPS_SamplerRequestHandle *req_handle;
79 * The task for this function.
81 struct GNUNET_SCHEDULER_Task *get_peer_task;
86 RPS_sampler_rand_peer_ready_cont cont;
89 * The closure to the callback @e cont
94 * The address of the id to be stored at
96 struct GNUNET_PeerIdentity *id;
101 * Type of function used to differentiate between modified and not modified
105 (*RPS_get_peers_type) (void *cls);
108 * Get one random peer out of the sampled peers.
110 * We might want to reinitialise this sampler after giving the
111 * corrsponding peer to the client.
112 * Only used internally
115 sampler_get_rand_peer (void *cls);
119 * Get one random peer out of the sampled peers.
121 * We might want to reinitialise this sampler after giving the
122 * corrsponding peer to the client.
125 sampler_mod_get_rand_peer (void *cls);
129 * Sampler with its own array of SamplerElements
134 * Number of sampler elements we hold.
136 unsigned int sampler_size;
140 * All sampler elements in one array.
142 struct RPS_SamplerElement **sampler_elements;
145 * Maximum time a round takes
147 * Used in the context of RPS
149 struct GNUNET_TIME_Relative max_round_interval;
152 * Stores the function to return peers. Which one it is depends on whether
153 * the Sampler is the modified one or not.
155 RPS_get_peers_type get_peers;
158 * Head and tail for the DLL to store the #RPS_SamplerRequestHandle
160 struct RPS_SamplerRequestHandle *req_handle_head;
161 struct RPS_SamplerRequestHandle *req_handle_tail;
165 * File name to log to
172 * Closure to _get_n_rand_peers_ready_cb()
174 struct RPS_SamplerRequestHandle
179 struct RPS_SamplerRequestHandle *next;
180 struct RPS_SamplerRequestHandle *prev;
183 * Number of peers we are waiting for.
188 * Number of peers we currently have.
190 uint32_t cur_num_peers;
193 * Pointer to the array holding the ids.
195 struct GNUNET_PeerIdentity *ids;
198 * Head and tail for the DLL to store the tasks for single requests
200 struct GetPeerCls *gpc_head;
201 struct GetPeerCls *gpc_tail;
206 struct RPS_Sampler *sampler;
209 * Callback to be called when all ids are available.
211 RPS_sampler_n_rand_peers_ready_cb callback;
214 * Closure given to the callback
220 // * Global sampler variable.
222 //struct RPS_Sampler *sampler;
226 * The minimal size for the extended sampler elements.
228 static size_t min_size;
231 * The maximal size the extended sampler elements should grow to.
233 static size_t max_size;
236 * The size the extended sampler elements currently have.
238 //static size_t extra_size;
241 * Inedex to the sampler element that is the next to be returned
243 static uint32_t client_get_index;
247 * Callback to _get_rand_peer() used by _get_n_rand_peers().
249 * Checks whether all n peers are available. If they are,
253 check_n_peers_ready (void *cls,
254 const struct GNUNET_PeerIdentity *id)
256 struct RPS_SamplerRequestHandle *req_handle = cls;
258 req_handle->cur_num_peers++;
259 LOG (GNUNET_ERROR_TYPE_DEBUG,
260 "Got %" PRIX32 ". of %" PRIX32 " peers\n",
261 req_handle->cur_num_peers, req_handle->num_peers);
263 if (req_handle->num_peers == req_handle->cur_num_peers)
264 { /* All peers are ready -- return those to the client */
265 GNUNET_assert (NULL != req_handle->callback);
267 LOG (GNUNET_ERROR_TYPE_DEBUG,
268 "returning %" PRIX32 " peers to the client\n",
269 req_handle->num_peers);
270 req_handle->callback (req_handle->cls, req_handle->ids, req_handle->num_peers);
272 RPS_sampler_request_cancel (req_handle);
278 * Get the size of the sampler.
280 * @param sampler the sampler to return the size of.
281 * @return the size of the sampler
284 RPS_sampler_get_size (struct RPS_Sampler *sampler)
286 return sampler->sampler_size;
291 * Grow or shrink the size of the sampler.
293 * @param sampler the sampler to resize.
294 * @param new_size the new size of the sampler
297 sampler_resize (struct RPS_Sampler *sampler, unsigned int new_size)
299 unsigned int old_size;
302 // TODO check min and max size
304 old_size = sampler->sampler_size;
306 if (old_size > new_size)
309 LOG (GNUNET_ERROR_TYPE_DEBUG,
310 "Shrinking sampler %d -> %d\n",
314 to_file (sampler->file_name,
315 "Shrinking sampler %d -> %d",
319 for (i = new_size ; i < old_size ; i++)
321 to_file (sampler->file_name,
324 sampler->sampler_elements[i]->file_name);
325 RPS_sampler_elem_destroy (sampler->sampler_elements[i]);
328 GNUNET_array_grow (sampler->sampler_elements,
329 sampler->sampler_size,
331 LOG (GNUNET_ERROR_TYPE_DEBUG,
332 "sampler->sampler_elements now points to %p\n",
333 sampler->sampler_elements);
336 else if (old_size < new_size)
338 LOG (GNUNET_ERROR_TYPE_DEBUG,
339 "Growing sampler %d -> %d\n",
343 to_file (sampler->file_name,
344 "Growing sampler %d -> %d",
348 GNUNET_array_grow (sampler->sampler_elements,
349 sampler->sampler_size,
352 for (i = old_size ; i < new_size ; i++)
353 { /* Add new sampler elements */
354 sampler->sampler_elements[i] = RPS_sampler_elem_create ();
356 to_file (sampler->file_name,
359 sampler->sampler_elements[i]->file_name);
364 LOG (GNUNET_ERROR_TYPE_DEBUG, "Size remains the same -- nothing to do\n");
368 GNUNET_assert (sampler->sampler_size == new_size);
373 * Grow or shrink the size of the sampler.
375 * @param sampler the sampler to resize.
376 * @param new_size the new size of the sampler
379 RPS_sampler_resize (struct RPS_Sampler *sampler, unsigned int new_size)
381 GNUNET_assert (0 < new_size);
382 sampler_resize (sampler, new_size);
389 * @param sampler the sampler to empty.
390 * @param new_size the new size of the sampler
393 sampler_empty (struct RPS_Sampler *sampler)
395 sampler_resize (sampler, 0);
400 * Initialise a tuple of sampler elements.
402 * @param init_size the size the sampler is initialised with
403 * @param max_round_interval maximum time a round takes
404 * @return a handle to a sampler that consists of sampler elements.
407 RPS_sampler_init (size_t init_size,
408 struct GNUNET_TIME_Relative max_round_interval)
410 struct RPS_Sampler *sampler;
412 /* Initialise context around extended sampler */
413 min_size = 10; // TODO make input to _samplers_init()
414 max_size = 1000; // TODO make input to _samplers_init()
416 sampler = GNUNET_new (struct RPS_Sampler);
419 sampler->file_name = create_file ("sampler-");
421 LOG (GNUNET_ERROR_TYPE_DEBUG,
422 "Initialised sampler %s\n",
426 sampler->max_round_interval = max_round_interval;
427 sampler->get_peers = sampler_get_rand_peer;
428 //sampler->sampler_elements = GNUNET_new_array(init_size, struct GNUNET_PeerIdentity);
429 //GNUNET_array_grow (sampler->sampler_elements, sampler->sampler_size, min_size);
430 RPS_sampler_resize (sampler, init_size);
432 client_get_index = 0;
434 //GNUNET_assert (init_size == sampler->sampler_size);
439 * Initialise a modified tuple of sampler elements.
441 * @param init_size the size the sampler is initialised with
442 * @param max_round_interval maximum time a round takes
443 * @return a handle to a sampler that consists of sampler elements.
446 RPS_sampler_mod_init (size_t init_size,
447 struct GNUNET_TIME_Relative max_round_interval)
449 struct RPS_Sampler *sampler;
451 sampler = RPS_sampler_init (init_size, max_round_interval);
452 sampler->get_peers = sampler_mod_get_rand_peer;
454 LOG (GNUNET_ERROR_TYPE_DEBUG,
455 "Initialised modified sampler %s\n",
457 to_file (sampler->file_name,
458 "This is a modified sampler");
465 * A fuction to update every sampler in the given list
467 * @param sampler the sampler to update.
468 * @param id the PeerID that is put in the sampler
471 RPS_sampler_update (struct RPS_Sampler *sampler,
472 const struct GNUNET_PeerIdentity *id)
476 to_file (sampler->file_name,
478 GNUNET_i2s_full (id));
480 for (i = 0 ; i < sampler->sampler_size ; i++)
482 RPS_sampler_elem_next (sampler->sampler_elements[i],
490 * Reinitialise all previously initialised sampler elements with the given value.
492 * Used to get rid of a PeerID.
494 * @param sampler the sampler to reinitialise a sampler element in.
495 * @param id the id of the sampler elements to update.
498 RPS_sampler_reinitialise_by_value (struct RPS_Sampler *sampler,
499 const struct GNUNET_PeerIdentity *id)
503 for (i = 0; i < sampler->sampler_size; i++)
505 if (0 == GNUNET_CRYPTO_cmp_peer_identity(id,
506 &(sampler->sampler_elements[i]->peer_id)) )
508 LOG (GNUNET_ERROR_TYPE_DEBUG, "Reinitialising sampler\n");
509 to_file (sampler->sampler_elements[i]->file_name,
511 RPS_sampler_elem_reinit (sampler->sampler_elements[i]);
518 * Get one random peer out of the sampled peers.
520 * We might want to reinitialise this sampler after giving the
521 * corrsponding peer to the client.
522 * Only used internally
525 sampler_get_rand_peer (void *cls)
527 struct GetPeerCls *gpc = cls;
529 struct RPS_Sampler *sampler;
531 gpc->get_peer_task = NULL;
532 sampler = gpc->req_handle->sampler;
535 * Choose the r_index of the peer we want to return
536 * at random from the interval of the gossip list
538 r_index = GNUNET_CRYPTO_random_u64 (GNUNET_CRYPTO_QUALITY_STRONG,
539 sampler->sampler_size);
541 if (EMPTY == sampler->sampler_elements[r_index]->is_empty)
543 //LOG (GNUNET_ERROR_TYPE_DEBUG,
544 // "Not returning randomly selected, empty PeerID. - Rescheduling.\n");
546 /* FIXME no active wait - get notified, when new id arrives?
547 * Might also be a freshly emptied one. Others might still contain ids.
551 GNUNET_SCHEDULER_add_delayed (
552 GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 1),
553 &sampler_get_rand_peer,
558 GNUNET_CONTAINER_DLL_remove (gpc->req_handle->gpc_head,
559 gpc->req_handle->gpc_tail,
561 *gpc->id = sampler->sampler_elements[r_index]->peer_id;
562 gpc->cont (gpc->cont_cls, gpc->id);
569 * Get one random peer out of the sampled peers.
571 * We might want to reinitialise this sampler after giving the
572 * corrsponding peer to the client.
575 sampler_mod_get_rand_peer (void *cls)
577 struct GetPeerCls *gpc = cls;
578 struct RPS_SamplerElement *s_elem;
579 struct GNUNET_TIME_Relative last_request_diff;
580 struct RPS_Sampler *sampler;
582 gpc->get_peer_task = NULL;
583 sampler = gpc->req_handle->sampler;
585 LOG (GNUNET_ERROR_TYPE_DEBUG, "Single peer was requested\n");
587 /* Cycle the #client_get_index one step further */
588 client_get_index = (client_get_index + 1) % sampler->sampler_size;
590 s_elem = sampler->sampler_elements[client_get_index];
591 *gpc->id = s_elem->peer_id;
592 GNUNET_assert (NULL != s_elem);
594 if (EMPTY == s_elem->is_empty)
596 LOG (GNUNET_ERROR_TYPE_DEBUG,
597 "Sampler_mod element empty, rescheduling.\n");
598 GNUNET_assert (NULL == gpc->get_peer_task);
600 GNUNET_SCHEDULER_add_delayed (sampler->max_round_interval,
601 &sampler_mod_get_rand_peer,
606 /* Check whether we may use this sampler to give it back to the client */
607 if (GNUNET_TIME_UNIT_FOREVER_ABS.abs_value_us != s_elem->last_client_request.abs_value_us)
610 GNUNET_TIME_absolute_get_difference (s_elem->last_client_request,
611 GNUNET_TIME_absolute_get ());
612 /* We're not going to give it back now if it was
613 * already requested by a client this round */
614 if (last_request_diff.rel_value_us < sampler->max_round_interval.rel_value_us)
616 LOG (GNUNET_ERROR_TYPE_DEBUG,
617 "Last client request on this sampler was less than max round interval ago -- scheduling for later\n");
618 ///* How many time remains untile the next round has started? */
619 //inv_last_request_diff =
620 // GNUNET_TIME_absolute_get_difference (last_request_diff,
621 // sampler->max_round_interval);
622 // add a little delay
623 /* Schedule it one round later */
624 GNUNET_assert (NULL == gpc->get_peer_task);
626 GNUNET_SCHEDULER_add_delayed (sampler->max_round_interval,
627 &sampler_mod_get_rand_peer,
631 // TODO add other reasons to wait here
634 s_elem->last_client_request = GNUNET_TIME_absolute_get ();
636 GNUNET_CONTAINER_DLL_remove (gpc->req_handle->gpc_head,
637 gpc->req_handle->gpc_tail,
639 gpc->cont (gpc->cont_cls, gpc->id);
645 * Get n random peers out of the sampled peers.
647 * We might want to reinitialise this sampler after giving the
648 * corrsponding peer to the client.
649 * Random with or without consumption?
651 * @param sampler the sampler to get peers from.
652 * @param cb callback that will be called once the ids are ready.
653 * @param cls closure given to @a cb
654 * @param for_client #GNUNET_YES if result is used for client,
655 * #GNUNET_NO if used internally
656 * @param num_peers the number of peers requested
658 struct RPS_SamplerRequestHandle *
659 RPS_sampler_get_n_rand_peers (struct RPS_Sampler *sampler,
660 RPS_sampler_n_rand_peers_ready_cb cb,
661 void *cls, uint32_t num_peers)
663 GNUNET_assert (0 != sampler->sampler_size);
667 // TODO check if we have too much (distinct) sampled peers
669 struct RPS_SamplerRequestHandle *req_handle;
670 struct GetPeerCls *gpc;
672 req_handle = GNUNET_new (struct RPS_SamplerRequestHandle);
673 req_handle->num_peers = num_peers;
674 req_handle->cur_num_peers = 0;
675 req_handle->ids = GNUNET_new_array (num_peers, struct GNUNET_PeerIdentity);
676 req_handle->sampler = sampler;
677 req_handle->callback = cb;
678 req_handle->cls = cls;
679 GNUNET_CONTAINER_DLL_insert (sampler->req_handle_head,
680 sampler->req_handle_tail,
683 LOG (GNUNET_ERROR_TYPE_DEBUG,
684 "Scheduling requests for %" PRIu32 " peers\n", num_peers);
686 for (i = 0 ; i < num_peers ; i++)
688 gpc = GNUNET_new (struct GetPeerCls);
689 gpc->req_handle = req_handle;
690 gpc->cont = check_n_peers_ready;
691 gpc->cont_cls = req_handle;
692 gpc->id = &req_handle->ids[i];
694 GNUNET_CONTAINER_DLL_insert (req_handle->gpc_head,
695 req_handle->gpc_tail,
697 // maybe add a little delay
698 gpc->get_peer_task = GNUNET_SCHEDULER_add_now (sampler->get_peers,
705 * Cancle a request issued through #RPS_sampler_n_rand_peers_ready_cb.
707 * @param req_handle the handle to the request
710 RPS_sampler_request_cancel (struct RPS_SamplerRequestHandle *req_handle)
712 struct GetPeerCls *i;
714 while (NULL != (i = req_handle->gpc_head) )
716 GNUNET_CONTAINER_DLL_remove (req_handle->gpc_head,
717 req_handle->gpc_tail,
719 if (NULL != i->get_peer_task)
721 GNUNET_SCHEDULER_cancel (i->get_peer_task);
725 GNUNET_CONTAINER_DLL_remove (req_handle->sampler->req_handle_head,
726 req_handle->sampler->req_handle_tail,
728 GNUNET_free (req_handle);
733 * Counts how many Samplers currently hold a given PeerID.
735 * @param sampler the sampler to count ids in.
736 * @param id the PeerID to count.
738 * @return the number of occurrences of id.
741 RPS_sampler_count_id (struct RPS_Sampler *sampler,
742 const struct GNUNET_PeerIdentity *id)
748 for ( i = 0 ; i < sampler->sampler_size ; i++ )
750 if ( 0 == GNUNET_CRYPTO_cmp_peer_identity (&sampler->sampler_elements[i]->peer_id, id)
751 && EMPTY != sampler->sampler_elements[i]->is_empty)
759 * Cleans the sampler.
762 RPS_sampler_destroy (struct RPS_Sampler *sampler)
764 if (NULL != sampler->req_handle_head)
766 LOG (GNUNET_ERROR_TYPE_WARNING,
767 "There are still pending requests. Going to remove them.\n");
768 while (NULL != sampler->req_handle_head)
770 RPS_sampler_request_cancel (sampler->req_handle_head);
773 sampler_empty (sampler);
774 GNUNET_free (sampler);
777 /* end of gnunet-service-rps.c */