aboutsummaryrefslogtreecommitdiff
path: root/src/rps/rps-sampler_common.c
diff options
context:
space:
mode:
Diffstat (limited to 'src/rps/rps-sampler_common.c')
-rw-r--r--src/rps/rps-sampler_common.c764
1 files changed, 0 insertions, 764 deletions
diff --git a/src/rps/rps-sampler_common.c b/src/rps/rps-sampler_common.c
deleted file mode 100644
index a30ba06a1..000000000
--- a/src/rps/rps-sampler_common.c
+++ /dev/null
@@ -1,764 +0,0 @@
1/*
2 This file is part of GNUnet.
3 Copyright (C)
4
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.
9
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.
14
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/>.
17
18 SPDX-License-Identifier: AGPL3.0-or-later
19 */
20
21/**
22 * @file rps/rps-sampler_common.c
23 * @brief Code common to client and service sampler
24 * @author Julius Bünger
25 */
26#include "platform.h"
27#include "gnunet_util_lib.h"
28#include "gnunet_statistics_service.h"
29
30#include "rps-sampler_common.h"
31#include "gnunet-service-rps_sampler_elem.h"
32
33#include <math.h>
34#include <inttypes.h>
35
36#include "rps-test_util.h"
37
38#define LOG(kind, ...) GNUNET_log_from (kind, "rps-sampler_common", __VA_ARGS__)
39
40/**
41 * @brief Context for a callback. Contains callback and closure.
42 *
43 * Meant to be an entry in an DLL.
44 */
45struct SamplerNotifyUpdateCTX
46{
47 /**
48 * @brief The Callback to call on updates
49 */
50 SamplerNotifyUpdateCB notify_cb;
51
52 /**
53 * @brief The according closure.
54 */
55 void *cls;
56
57 /**
58 * @brief Next element in DLL.
59 */
60 struct SamplerNotifyUpdateCTX *next;
61
62 /**
63 * @brief Previous element in DLL.
64 */
65 struct SamplerNotifyUpdateCTX *prev;
66};
67
68
69/**
70 * Closure to _get_n_rand_peers_ready_cb()
71 */
72struct RPS_SamplerRequestHandle
73{
74 /**
75 * DLL
76 */
77 struct RPS_SamplerRequestHandle *next;
78 struct RPS_SamplerRequestHandle *prev;
79
80 /**
81 * Number of peers we are waiting for.
82 */
83 uint32_t num_peers;
84
85 /**
86 * Number of peers we currently have.
87 */
88 uint32_t cur_num_peers;
89
90 /**
91 * Pointer to the array holding the ids.
92 */
93 struct GNUNET_PeerIdentity *ids;
94
95 /**
96 * Head and tail for the DLL to store the tasks for single requests
97 */
98 struct GetPeerCls *gpc_head;
99 struct GetPeerCls *gpc_tail;
100
101 /**
102 * Sampler.
103 */
104 struct RPS_Sampler *sampler;
105
106 /**
107 * Callback to be called when all ids are available.
108 */
109 RPS_sampler_n_rand_peers_ready_cb callback;
110
111 /**
112 * Closure given to the callback
113 */
114 void *cls;
115};
116
117
118/**
119 * Closure to _get_rand_peer_info()
120 */
121struct RPS_SamplerRequestHandleSingleInfo
122{
123 /**
124 * DLL
125 */
126 struct RPS_SamplerRequestHandleSingleInfo *next;
127 struct RPS_SamplerRequestHandleSingleInfo *prev;
128
129 /**
130 * Pointer to the id
131 */
132 struct GNUNET_PeerIdentity *id;
133
134 /**
135 * Head and tail for the DLL to store the tasks for single requests
136 */
137 struct GetPeerCls *gpc_head;
138 struct GetPeerCls *gpc_tail;
139
140 /**
141 * Sampler.
142 */
143 struct RPS_Sampler *sampler;
144
145 /**
146 * Callback to be called when all ids are available.
147 */
148 RPS_sampler_sinlge_info_ready_cb callback;
149
150 /**
151 * Closure given to the callback
152 */
153 void *cls;
154};
155
156
157/**
158 * @brief Update the current estimate of the network size stored at the sampler
159 *
160 * Used for computing the condition when to return elements to the client
161 *
162 * Only used/useful with the client sampler
163 * (Maybe move to rps-sampler_client.{h|c} ?)
164 *
165 * @param sampler The sampler to update
166 * @param num_peers The estimated value
167 */
168void
169RPS_sampler_update_with_nw_size (struct RPS_Sampler *sampler,
170 uint32_t num_peers)
171{
172 sampler->num_peers_estim = num_peers;
173}
174
175
176/**
177 * @brief Set the probability that is needed at least with what a sampler
178 * element has to have observed all elements from the network.
179 *
180 * Only used/useful with the client sampler
181 * (Maybe move to rps-sampler_client.{h|c} ?)
182 *
183 * @param sampler
184 * @param desired_probability
185 */
186void
187RPS_sampler_set_desired_probability (struct RPS_Sampler *sampler,
188 double desired_probability)
189{
190 sampler->desired_probability = desired_probability;
191}
192
193
194/**
195 * @brief Set the deficiency factor.
196 *
197 * Only used/useful with the client sampler
198 * (Maybe move to rps-sampler_client.{h|c} ?)
199 *
200 * @param sampler
201 * @param desired_probability
202 */
203void
204RPS_sampler_set_deficiency_factor (struct RPS_Sampler *sampler,
205 double deficiency_factor)
206{
207 sampler->deficiency_factor = deficiency_factor;
208}
209
210
211/**
212 * @brief Add a callback that will be called when the next peer is inserted
213 * into the sampler
214 *
215 * @param sampler The sampler on which update it will be called
216 * @param notify_cb The callback
217 * @param cls Closure given to the callback
218 *
219 * @return The context containing callback and closure
220 */
221struct SamplerNotifyUpdateCTX *
222sampler_notify_on_update (struct RPS_Sampler *sampler,
223 SamplerNotifyUpdateCB notify_cb,
224 void *cls)
225{
226 struct SamplerNotifyUpdateCTX *notify_ctx;
227
228 LOG (GNUNET_ERROR_TYPE_DEBUG,
229 "Inserting new context for notification\n");
230 notify_ctx = GNUNET_new (struct SamplerNotifyUpdateCTX);
231 notify_ctx->notify_cb = notify_cb;
232 notify_ctx->cls = cls;
233 GNUNET_CONTAINER_DLL_insert (sampler->notify_ctx_head,
234 sampler->notify_ctx_tail,
235 notify_ctx);
236 return notify_ctx;
237}
238
239
240/**
241 * Get the size of the sampler.
242 *
243 * @param sampler the sampler to return the size of.
244 * @return the size of the sampler
245 */
246unsigned int
247RPS_sampler_get_size (struct RPS_Sampler *sampler)
248{
249 return sampler->sampler_size;
250}
251
252
253/**
254 * @brief Notify about update of the sampler.
255 *
256 * Call the callbacks that are waiting for notification on updates to the
257 * sampler.
258 *
259 * @param sampler The sampler the updates are waiting for
260 */
261static void
262notify_update (struct RPS_Sampler *sampler)
263{
264 struct SamplerNotifyUpdateCTX *tmp_notify_head;
265 struct SamplerNotifyUpdateCTX *tmp_notify_tail;
266
267 LOG (GNUNET_ERROR_TYPE_DEBUG,
268 "Calling callbacks waiting for update notification.\n");
269 tmp_notify_head = sampler->notify_ctx_head;
270 tmp_notify_tail = sampler->notify_ctx_tail;
271 sampler->notify_ctx_head = NULL;
272 sampler->notify_ctx_tail = NULL;
273 for (struct SamplerNotifyUpdateCTX *notify_iter = tmp_notify_head;
274 NULL != tmp_notify_head;
275 notify_iter = tmp_notify_head)
276 {
277 GNUNET_assert (NULL != notify_iter->notify_cb);
278 GNUNET_CONTAINER_DLL_remove (tmp_notify_head,
279 tmp_notify_tail,
280 notify_iter);
281 notify_iter->notify_cb (notify_iter->cls);
282 GNUNET_free (notify_iter);
283 }
284}
285
286
287/**
288 * Update every sampler element of this sampler with given peer
289 *
290 * @param sampler the sampler to update.
291 * @param id the PeerID that is put in the sampler
292 */
293void
294RPS_sampler_update (struct RPS_Sampler *sampler,
295 const struct GNUNET_PeerIdentity *id)
296{
297 for (uint32_t i = 0; i < sampler->sampler_size; i++)
298 {
299 RPS_sampler_elem_next (sampler->sampler_elements[i],
300 id);
301 }
302 notify_update (sampler);
303}
304
305
306/**
307 * Reinitialise all previously initialised sampler elements with the given value.
308 *
309 * Used to get rid of a PeerID.
310 *
311 * FIXME: This should also consider currently pending requests
312 * (Pending requests already collect peerids. As long as not all
313 * requested IDs have been collected, they are kept.
314 * Ideally, the @p id should be removed from all pending requests. This
315 * seems quite complicated.)
316 *
317 * @param sampler the sampler to reinitialise a sampler element in.
318 * @param id the id of the sampler elements to update.
319 */
320void
321RPS_sampler_reinitialise_by_value (struct RPS_Sampler *sampler,
322 const struct GNUNET_PeerIdentity *id)
323{
324 uint32_t i;
325
326 for (i = 0; i < sampler->sampler_size; i++)
327 {
328 if (0 == GNUNET_memcmp (id,
329 &(sampler->sampler_elements[i]->peer_id)))
330 {
331 LOG (GNUNET_ERROR_TYPE_DEBUG, "Reinitialising sampler\n");
332 RPS_sampler_elem_reinit (sampler->sampler_elements[i]);
333 }
334 }
335}
336
337
338/**
339 * Counts how many Samplers currently hold a given PeerID.
340 *
341 * @param sampler the sampler to count ids in.
342 * @param id the PeerID to count.
343 *
344 * @return the number of occurrences of id.
345 */
346uint32_t
347RPS_sampler_count_id (struct RPS_Sampler *sampler,
348 const struct GNUNET_PeerIdentity *id)
349{
350 uint32_t count;
351 uint32_t i;
352
353 count = 0;
354 for (i = 0; i < sampler->sampler_size; i++)
355 {
356 if ((0 == GNUNET_memcmp (&sampler->sampler_elements[i]->peer_id, id))
357 && (EMPTY != sampler->sampler_elements[i]->is_empty) )
358 count++;
359 }
360 return count;
361}
362
363
364/**
365 * Grow or shrink the size of the sampler.
366 *
367 * @param sampler the sampler to resize.
368 * @param new_size the new size of the sampler
369 */
370static void
371sampler_resize (struct RPS_Sampler *sampler, unsigned int new_size)
372{
373 unsigned int old_size;
374 uint32_t i;
375
376 // TODO check min and max size
377
378 old_size = sampler->sampler_size;
379
380 if (old_size > new_size)
381 { /* Shrinking */
382 LOG (GNUNET_ERROR_TYPE_DEBUG,
383 "Shrinking sampler %d -> %d\n",
384 old_size,
385 new_size);
386
387 for (i = new_size; i < old_size; i++)
388 {
389 RPS_sampler_elem_destroy (sampler->sampler_elements[i]);
390 }
391
392 GNUNET_array_grow (sampler->sampler_elements,
393 sampler->sampler_size,
394 new_size);
395 LOG (GNUNET_ERROR_TYPE_DEBUG,
396 "sampler->sampler_elements now points to %p\n",
397 sampler->sampler_elements);
398 }
399 else if (old_size < new_size)
400 { /* Growing */
401 LOG (GNUNET_ERROR_TYPE_DEBUG,
402 "Growing sampler %d -> %d\n",
403 old_size,
404 new_size);
405
406 GNUNET_array_grow (sampler->sampler_elements,
407 sampler->sampler_size,
408 new_size);
409
410 for (i = old_size; i < new_size; i++)
411 { /* Add new sampler elements */
412 sampler->sampler_elements[i] = RPS_sampler_elem_create ();
413 }
414 }
415 else
416 {
417 LOG (GNUNET_ERROR_TYPE_DEBUG, "Size remains the same -- nothing to do\n");
418 return;
419 }
420
421 GNUNET_assert (sampler->sampler_size == new_size);
422}
423
424
425/**
426 * Grow or shrink the size of the sampler.
427 *
428 * @param sampler the sampler to resize.
429 * @param new_size the new size of the sampler
430 */
431void
432RPS_sampler_resize (struct RPS_Sampler *sampler, unsigned int new_size)
433{
434 GNUNET_assert (0 < new_size);
435 sampler_resize (sampler, new_size);
436}
437
438
439/**
440 * Empty the sampler.
441 *
442 * @param sampler the sampler to empty.
443 * @param new_size the new size of the sampler
444 */
445static void
446sampler_empty (struct RPS_Sampler *sampler)
447{
448 sampler_resize (sampler, 0);
449}
450
451
452/**
453 * Callback to _get_rand_peer() used by _get_n_rand_peers().
454 *
455 * Implements #RPS_sampler_rand_peer_ready_cont
456 *
457 * Checks whether all n peers are available. If they are,
458 * give those back.
459 * @param cls Closure
460 * @param id Peer ID
461 * @param probability The probability with which this sampler has seen all ids
462 * @param num_observed How many ids this sampler has observed
463 */
464static void
465check_n_peers_ready (void *cls,
466 const struct GNUNET_PeerIdentity *id,
467 double probability,
468 uint32_t num_observed)
469{
470 struct RPS_SamplerRequestHandle *req_handle = cls;
471
472 (void) id;
473 RPS_sampler_n_rand_peers_ready_cb tmp_cb;
474 struct GNUNET_PeerIdentity *peers;
475 uint32_t num_peers;
476 void *cb_cls;
477 (void) probability;
478 (void) num_observed;
479
480 req_handle->cur_num_peers++;
481 LOG (GNUNET_ERROR_TYPE_DEBUG,
482 "Got %" PRIX32 ". of %" PRIX32 " peers\n",
483 req_handle->cur_num_peers, req_handle->num_peers);
484
485 if (req_handle->num_peers == req_handle->cur_num_peers)
486 { /* All peers are ready -- return those to the client */
487 GNUNET_assert (NULL != req_handle->callback);
488
489 LOG (GNUNET_ERROR_TYPE_DEBUG,
490 "returning %" PRIX32 " peers to the client\n",
491 req_handle->num_peers);
492
493 /* Copy pointers and peers temporarily as they
494 * might be deleted from within the callback */
495 tmp_cb = req_handle->callback;
496 num_peers = req_handle->num_peers;
497 peers = GNUNET_new_array (num_peers, struct GNUNET_PeerIdentity);
498 GNUNET_memcpy (peers,
499 req_handle->ids,
500 num_peers * sizeof(struct GNUNET_PeerIdentity));
501 cb_cls = req_handle->cls;
502 RPS_sampler_request_cancel (req_handle);
503 req_handle = NULL;
504 tmp_cb (peers, num_peers, cb_cls);
505 GNUNET_free (peers);
506 }
507}
508
509
510/**
511 * Callback to _get_rand_peer() used by _get_rand_peer_info().
512 *
513 * Implements #RPS_sampler_rand_peer_ready_cont
514 *
515 * @param cls Closure
516 * @param id Peer ID
517 * @param probability The probability with which this sampler has seen all ids
518 * @param num_observed How many ids this sampler has observed
519 */
520static void
521check_peer_info_ready (void *cls,
522 const struct GNUNET_PeerIdentity *id,
523 double probability,
524 uint32_t num_observed)
525{
526 struct RPS_SamplerRequestHandleSingleInfo *req_handle = cls;
527
528 (void) id;
529 RPS_sampler_sinlge_info_ready_cb tmp_cb;
530 struct GNUNET_PeerIdentity *peer;
531 void *cb_cls;
532 (void) probability;
533 (void) num_observed;
534
535 LOG (GNUNET_ERROR_TYPE_DEBUG,
536 "Got single peer with additional info\n");
537
538 GNUNET_assert (NULL != req_handle->callback);
539
540 LOG (GNUNET_ERROR_TYPE_DEBUG,
541 "returning single peer with info to the client\n");
542
543 /* Copy pointers and peers temporarily as they
544 * might be deleted from within the callback */
545 tmp_cb = req_handle->callback;
546 peer = GNUNET_new (struct GNUNET_PeerIdentity);
547 GNUNET_memcpy (peer,
548 req_handle->id,
549 sizeof(struct GNUNET_PeerIdentity));
550 cb_cls = req_handle->cls;
551 RPS_sampler_request_single_info_cancel (req_handle);
552 req_handle = NULL;
553 tmp_cb (peer, cb_cls, probability, num_observed);
554 GNUNET_free (peer);
555}
556
557
558/**
559 * Get n random peers out of the sampled peers.
560 *
561 * We might want to reinitialise this sampler after giving the
562 * corrsponding peer to the client.
563 * Random with or without consumption?
564 *
565 * @param sampler the sampler to get peers from.
566 * @param cb callback that will be called once the ids are ready.
567 * @param cls closure given to @a cb
568 * @param num_peers the number of peers requested
569 */
570struct RPS_SamplerRequestHandle *
571RPS_sampler_get_n_rand_peers (struct RPS_Sampler *sampler,
572 uint32_t num_peers,
573 RPS_sampler_n_rand_peers_ready_cb cb,
574 void *cls)
575{
576 uint32_t i;
577 struct RPS_SamplerRequestHandle *req_handle;
578 struct GetPeerCls *gpc;
579
580 GNUNET_assert (0 != sampler->sampler_size);
581 if (0 == num_peers)
582 return NULL;
583
584 // TODO check if we have too much (distinct) sampled peers
585 req_handle = GNUNET_new (struct RPS_SamplerRequestHandle);
586 req_handle->num_peers = num_peers;
587 req_handle->cur_num_peers = 0;
588 req_handle->ids = GNUNET_new_array (num_peers, struct GNUNET_PeerIdentity);
589 req_handle->sampler = sampler;
590 req_handle->callback = cb;
591 req_handle->cls = cls;
592 GNUNET_CONTAINER_DLL_insert (sampler->req_handle_head,
593 sampler->req_handle_tail,
594 req_handle);
595
596 LOG (GNUNET_ERROR_TYPE_DEBUG,
597 "Scheduling requests for %" PRIu32 " peers\n", num_peers);
598
599 for (i = 0; i < num_peers; i++)
600 {
601 gpc = GNUNET_new (struct GetPeerCls);
602 gpc->req_handle = req_handle;
603 gpc->req_single_info_handle = NULL;
604 gpc->cont = check_n_peers_ready;
605 gpc->cont_cls = req_handle;
606 gpc->id = &req_handle->ids[i];
607
608 GNUNET_CONTAINER_DLL_insert (req_handle->gpc_head,
609 req_handle->gpc_tail,
610 gpc);
611 // maybe add a little delay
612 gpc->get_peer_task = GNUNET_SCHEDULER_add_now (sampler->get_peers,
613 gpc);
614 }
615 return req_handle;
616}
617
618
619/**
620 * Get one random peer with additional information.
621 *
622 * @param sampler the sampler to get peers from.
623 * @param cb callback that will be called once the ids are ready.
624 * @param cls closure given to @a cb
625 */
626struct RPS_SamplerRequestHandleSingleInfo *
627RPS_sampler_get_rand_peer_info (struct RPS_Sampler *sampler,
628 RPS_sampler_sinlge_info_ready_cb cb,
629 void *cls)
630{
631 struct RPS_SamplerRequestHandleSingleInfo *req_handle;
632 struct GetPeerCls *gpc;
633
634 GNUNET_assert (0 != sampler->sampler_size);
635
636 // TODO check if we have too much (distinct) sampled peers
637 req_handle = GNUNET_new (struct RPS_SamplerRequestHandleSingleInfo);
638 req_handle->id = GNUNET_malloc (sizeof(struct GNUNET_PeerIdentity));
639 req_handle->sampler = sampler;
640 req_handle->callback = cb;
641 req_handle->cls = cls;
642 GNUNET_CONTAINER_DLL_insert (sampler->req_handle_single_head,
643 sampler->req_handle_single_tail,
644 req_handle);
645
646 gpc = GNUNET_new (struct GetPeerCls);
647 gpc->req_handle = NULL;
648 gpc->req_single_info_handle = req_handle;
649 gpc->cont = check_peer_info_ready;
650 gpc->cont_cls = req_handle;
651 gpc->id = req_handle->id;
652
653 GNUNET_CONTAINER_DLL_insert (req_handle->gpc_head,
654 req_handle->gpc_tail,
655 gpc);
656 // maybe add a little delay
657 gpc->get_peer_task = GNUNET_SCHEDULER_add_now (sampler->get_peers,
658 gpc);
659 return req_handle;
660}
661
662
663/**
664 * Cancel a request issued through #RPS_sampler_n_rand_peers_ready_cb.
665 *
666 * @param req_handle the handle to the request
667 */
668void
669RPS_sampler_request_cancel (struct RPS_SamplerRequestHandle *req_handle)
670{
671 struct GetPeerCls *i;
672
673 while (NULL != (i = req_handle->gpc_head))
674 {
675 GNUNET_CONTAINER_DLL_remove (req_handle->gpc_head,
676 req_handle->gpc_tail,
677 i);
678 if (NULL != i->get_peer_task)
679 {
680 GNUNET_SCHEDULER_cancel (i->get_peer_task);
681 }
682 if (NULL != i->notify_ctx)
683 {
684 GNUNET_CONTAINER_DLL_remove (req_handle->sampler->notify_ctx_head,
685 req_handle->sampler->notify_ctx_tail,
686 i->notify_ctx);
687 GNUNET_free (i->notify_ctx);
688 i->notify_ctx = NULL;
689 }
690 GNUNET_free (i);
691 }
692 GNUNET_free (req_handle->ids);
693 req_handle->ids = NULL;
694 GNUNET_CONTAINER_DLL_remove (req_handle->sampler->req_handle_head,
695 req_handle->sampler->req_handle_tail,
696 req_handle);
697 GNUNET_free (req_handle);
698}
699
700
701/**
702 * Cancel a request issued through #RPS_sampler_sinlge_info_ready_cb.
703 *
704 * @param req_handle the handle to the request
705 */
706void
707RPS_sampler_request_single_info_cancel (
708 struct RPS_SamplerRequestHandleSingleInfo *req_single_info_handle)
709{
710 struct GetPeerCls *i;
711
712 while (NULL != (i = req_single_info_handle->gpc_head))
713 {
714 GNUNET_CONTAINER_DLL_remove (req_single_info_handle->gpc_head,
715 req_single_info_handle->gpc_tail,
716 i);
717 if (NULL != i->get_peer_task)
718 {
719 GNUNET_SCHEDULER_cancel (i->get_peer_task);
720 }
721 if (NULL != i->notify_ctx)
722 {
723 GNUNET_CONTAINER_DLL_remove (
724 req_single_info_handle->sampler->notify_ctx_head,
725 req_single_info_handle->sampler->
726 notify_ctx_tail,
727 i->notify_ctx);
728 GNUNET_free (i->notify_ctx);
729 i->notify_ctx = NULL;
730 }
731 GNUNET_free (i);
732 }
733 GNUNET_free (req_single_info_handle->id);
734 req_single_info_handle->id = NULL;
735 GNUNET_CONTAINER_DLL_remove (
736 req_single_info_handle->sampler->req_handle_single_head,
737 req_single_info_handle->sampler->
738 req_handle_single_tail,
739 req_single_info_handle);
740 GNUNET_free (req_single_info_handle);
741}
742
743
744/**
745 * Cleans the sampler.
746 */
747void
748RPS_sampler_destroy (struct RPS_Sampler *sampler)
749{
750 if (NULL != sampler->req_handle_head)
751 {
752 LOG (GNUNET_ERROR_TYPE_WARNING,
753 "There are still pending requests. Going to remove them.\n");
754 while (NULL != sampler->req_handle_head)
755 {
756 RPS_sampler_request_cancel (sampler->req_handle_head);
757 }
758 }
759 sampler_empty (sampler);
760 GNUNET_free (sampler);
761}
762
763
764/* end of rps-sampler_common.c */