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 Affero 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.
15 You should have received a copy of the GNU Affero General Public License
16 along with this program. If not, see <http://www.gnu.org/licenses/>.
18 SPDX-License-Identifier: AGPL3.0-or-later
22 * @file rps/gnunet-service-rps_sampler.c
23 * @brief sampler implementation
24 * @author Julius Bünger
27 #include "gnunet_util_lib.h"
28 #include "gnunet_statistics_service.h"
31 #include "rps-sampler_common.h"
32 #include "gnunet-service-rps_sampler.h"
33 #include "gnunet-service-rps_sampler_elem.h"
38 #include "rps-test_util.h"
40 #define LOG(kind, ...) GNUNET_log_from (kind, "rps-sampler", __VA_ARGS__)
43 // multiple 'clients'?
45 // TODO check for overflows
47 // TODO align message structs
49 // hist_size_init, hist_size_max
51 /***********************************************************************
52 * WARNING: This section needs to be reviewed regarding the use of
53 * functions providing (pseudo)randomness!
54 ***********************************************************************/
56 // TODO care about invalid input of the caller (size 0 or less...)
59 * @brief Callback called each time a new peer was put into the sampler
61 * @param cls A possibly given closure
64 (*SamplerNotifyUpdateCB) (void *cls);
67 * @brief Context for a callback. Contains callback and closure.
69 * Meant to be an entry in an DLL.
71 struct SamplerNotifyUpdateCTX
74 * @brief The Callback to call on updates
76 SamplerNotifyUpdateCB notify_cb;
79 * @brief The according closure.
84 * @brief Next element in DLL.
86 struct SamplerNotifyUpdateCTX *next;
89 * @brief Previous element in DLL.
91 struct SamplerNotifyUpdateCTX *prev;
96 * Type of function used to differentiate between modified and not modified
100 (*RPS_get_peers_type) (void *cls);
104 * Get one random peer out of the sampled peers.
106 * We might want to reinitialise this sampler after giving the
107 * corrsponding peer to the client.
110 sampler_mod_get_rand_peer (void *cls);
114 * Closure to _get_n_rand_peers_ready_cb()
116 struct RPS_SamplerRequestHandle
121 struct RPS_SamplerRequestHandle *next;
122 struct RPS_SamplerRequestHandle *prev;
125 * Number of peers we are waiting for.
130 * Number of peers we currently have.
132 uint32_t cur_num_peers;
135 * Pointer to the array holding the ids.
137 struct GNUNET_PeerIdentity *ids;
140 * Head and tail for the DLL to store the tasks for single requests
142 struct GetPeerCls *gpc_head;
143 struct GetPeerCls *gpc_tail;
148 struct RPS_Sampler *sampler;
151 * Callback to be called when all ids are available.
153 RPS_sampler_n_rand_peers_ready_cb callback;
156 * Closure given to the callback
163 * Closure to _get_rand_peer_info()
165 struct RPS_SamplerRequestHandleSingleInfo
170 struct RPS_SamplerRequestHandleSingleInfo *next;
171 struct RPS_SamplerRequestHandleSingleInfo *prev;
176 struct GNUNET_PeerIdentity *id;
179 * Head and tail for the DLL to store the tasks for single requests
181 struct GetPeerCls *gpc_head;
182 struct GetPeerCls *gpc_tail;
187 struct RPS_Sampler *sampler;
190 * Callback to be called when all ids are available.
192 RPS_sampler_sinlge_info_ready_cb callback;
195 * Closure given to the callback
202 // * Global sampler variable.
204 // struct RPS_Sampler *sampler;
208 * The minimal size for the extended sampler elements.
210 static size_t min_size;
213 * The maximal size the extended sampler elements should grow to.
215 static size_t max_size;
218 * The size the extended sampler elements currently have.
220 // static size_t extra_size;
223 * Inedex to the sampler element that is the next to be returned
225 static uint32_t client_get_index;
229 * Initialise a modified tuple of sampler elements.
231 * @param init_size the size the sampler is initialised with
232 * @param max_round_interval maximum time a round takes
233 * @return a handle to a sampler that consists of sampler elements.
236 RPS_sampler_mod_init (size_t init_size,
237 struct GNUNET_TIME_Relative max_round_interval)
239 struct RPS_Sampler *sampler;
241 /* Initialise context around extended sampler */
242 min_size = 10; // TODO make input to _samplers_init()
243 max_size = 1000; // TODO make input to _samplers_init()
245 sampler = GNUNET_new (struct RPS_Sampler);
246 sampler->max_round_interval = max_round_interval;
247 sampler->get_peers = sampler_mod_get_rand_peer;
248 // sampler->sampler_elements = GNUNET_new_array(init_size, struct GNUNET_PeerIdentity);
249 // GNUNET_array_grow (sampler->sampler_elements, sampler->sampler_size, min_size);
251 client_get_index = 0;
253 // GNUNET_assert (init_size == sampler->sampler_size);
255 RPS_sampler_resize (sampler, init_size);
262 * @brief Compute the probability that we already observed all peers from a
263 * biased stream of peer ids.
266 * As introduced by Brahms: Factor between the number of unique ids in a
267 * truly random stream and number of unique ids in the gossip stream.
269 * @param num_peers_estim The estimated number of peers in the network
270 * @param num_peers_observed The number of peers the given element has observed
271 * @param deficiency_factor A factor that catches the 'bias' of a random stream
274 * @return The estimated probability
277 prob_observed_n_peers (uint32_t num_peers_estim,
278 uint32_t num_peers_observed,
279 double deficiency_factor)
281 uint32_t num_peers = num_peers_estim * (1 / deficiency_factor);
284 for (uint32_t i = 0; i < num_peers; i++)
286 uint64_t a = pow (-1, num_peers - i);
287 uint64_t b = binom (num_peers, i);
288 uint64_t c = pow (i, num_peers_observed);
292 return sum / (double) pow (num_peers, num_peers_observed);
297 * Get one random peer out of the sampled peers.
299 * This reinitialises the queried sampler element.
302 sampler_mod_get_rand_peer (void *cls)
304 struct GetPeerCls *gpc = cls;
305 struct RPS_SamplerElement *s_elem;
306 struct GNUNET_TIME_Relative last_request_diff;
307 struct RPS_Sampler *sampler;
308 double prob_observed_n;
309 uint32_t num_observed;
311 gpc->get_peer_task = NULL;
312 gpc->notify_ctx = NULL;
313 GNUNET_assert ((NULL != gpc->req_handle) ||
314 (NULL != gpc->req_single_info_handle));
315 if (NULL != gpc->req_handle)
316 sampler = gpc->req_handle->sampler;
318 sampler = gpc->req_single_info_handle->sampler;
320 LOG (GNUNET_ERROR_TYPE_DEBUG, "Single peer was requested\n");
322 /* Cycle the #client_get_index one step further */
323 client_get_index = (client_get_index + 1) % sampler->sampler_size;
325 s_elem = sampler->sampler_elements[client_get_index];
326 *gpc->id = s_elem->peer_id;
327 GNUNET_assert (NULL != s_elem);
329 if (EMPTY == s_elem->is_empty)
331 LOG (GNUNET_ERROR_TYPE_DEBUG,
332 "Sampler_mod element empty, rescheduling.\n");
333 GNUNET_assert (NULL == gpc->notify_ctx);
335 sampler_notify_on_update (sampler,
336 &sampler_mod_get_rand_peer,
341 /* Check whether we may use this sampler to give it back to the client */
342 if (GNUNET_TIME_UNIT_FOREVER_ABS.abs_value_us !=
343 s_elem->last_client_request.abs_value_us)
345 // TODO remove this condition at least for the client sampler
347 GNUNET_TIME_absolute_get_difference (s_elem->last_client_request,
348 GNUNET_TIME_absolute_get ());
349 /* We're not going to give it back now if it was
350 * already requested by a client this round */
351 if (last_request_diff.rel_value_us <
352 sampler->max_round_interval.rel_value_us)
354 LOG (GNUNET_ERROR_TYPE_DEBUG,
355 "Last client request on this sampler was less than max round interval ago -- scheduling for later\n");
356 ///* How many time remains untile the next round has started? */
357 // inv_last_request_diff =
358 // GNUNET_TIME_absolute_get_difference (last_request_diff,
359 // sampler->max_round_interval);
360 // add a little delay
361 /* Schedule it one round later */
362 GNUNET_assert (NULL == gpc->notify_ctx);
364 sampler_notify_on_update (sampler,
365 &sampler_mod_get_rand_peer,
370 if (2 > s_elem->num_peers)
372 LOG (GNUNET_ERROR_TYPE_DEBUG,
373 "This s_elem saw less than two peers -- scheduling for later\n");
374 GNUNET_assert (NULL == gpc->notify_ctx);
376 sampler_notify_on_update (sampler,
377 &sampler_mod_get_rand_peer,
381 /* compute probability */
382 /* Currently disabled due to numerical limitations */
383 // prob_observed_n = prob_observed_n_peers (sampler->num_peers_estim,
384 // s_elem->num_peers,
385 // sampler->deficiency_factor);
386 // LOG (GNUNET_ERROR_TYPE_DEBUG,
387 // "Computed sample - prob %f, %" PRIu32 " peers, n: %" PRIu32 ", roh: %f\n",
389 // s_elem->num_peers,
390 // sampler->num_peers_estim,
391 // sampler->deficiency_factor);
392 ///* check if probability is above desired */
393 // if (prob_observed_n < sampler->desired_probability)
395 // LOG (GNUNET_ERROR_TYPE_DEBUG,
396 // "Probability of having observed all peers (%f) too small ( < %f).\n",
398 // sampler->desired_probability);
399 // GNUNET_assert (NULL == gpc->notify_ctx);
401 // sampler_notify_on_update (sampler,
402 // &sampler_mod_get_rand_peer,
406 /* More reasons to wait could be added here */
408 // GNUNET_STATISTICS_set (stats,
409 // "# client sampler element input",
410 // s_elem->num_peers,
412 // GNUNET_STATISTICS_set (stats,
413 // "# client sampler element change",
414 // s_elem->num_change,
417 num_observed = s_elem->num_peers;
418 RPS_sampler_elem_reinit (s_elem);
419 s_elem->last_client_request = GNUNET_TIME_absolute_get ();
421 if (NULL != gpc->req_handle)
423 GNUNET_CONTAINER_DLL_remove (gpc->req_handle->gpc_head,
424 gpc->req_handle->gpc_tail,
429 GNUNET_CONTAINER_DLL_remove (gpc->req_single_info_handle->gpc_head,
430 gpc->req_single_info_handle->gpc_tail,
433 gpc->cont (gpc->cont_cls, gpc->id, prob_observed_n, num_observed);
438 /* end of gnunet-service-rps.c */