paragraph for gnunet devs that don't know how to use the web
[oweals/gnunet.git] / src / rps / gnunet-service-rps_sampler.c
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
19 /**
20  * @file rps/gnunet-service-rps_sampler.c
21  * @brief sampler implementation
22  * @author Julius Bünger
23  */
24 #include "platform.h"
25 #include "gnunet_util_lib.h"
26 #include "rps.h"
27
28 #include "gnunet-service-rps_sampler.h"
29 #include "gnunet-service-rps_sampler_elem.h"
30
31 #include <math.h>
32 #include <inttypes.h>
33
34 #include "rps-test_util.h"
35
36 #define LOG(kind, ...) GNUNET_log_from(kind,"rps-sampler",__VA_ARGS__)
37
38
39 // multiple 'clients'?
40
41 // TODO check for overflows
42
43 // TODO align message structs
44
45 // hist_size_init, hist_size_max
46
47 /***********************************************************************
48  * WARNING: This section needs to be reviewed regarding the use of
49  * functions providing (pseudo)randomness!
50 ***********************************************************************/
51
52 // TODO care about invalid input of the caller (size 0 or less...)
53
54 /**
55  * Callback that is called from _get_rand_peer() when the PeerID is ready.
56  *
57  * @param cls the closure given alongside this function.
58  * @param id the PeerID that was returned
59  */
60 typedef void
61 (*RPS_sampler_rand_peer_ready_cont) (void *cls,
62                                      const struct GNUNET_PeerIdentity *id);
63
64
65 /**
66  * Closure for #sampler_mod_get_rand_peer() and #sampler_get_rand_peer
67  */
68 struct GetPeerCls
69 {
70   /**
71    * DLL
72    */
73   struct GetPeerCls *next;
74   struct GetPeerCls *prev;
75
76   /**
77    * The #RPS_SamplerRequestHandle this single request belongs to.
78    */
79   struct RPS_SamplerRequestHandle *req_handle;
80
81   /**
82    * The task for this function.
83    */
84   struct GNUNET_SCHEDULER_Task *get_peer_task;
85
86   /**
87    * The callback
88    */
89   RPS_sampler_rand_peer_ready_cont cont;
90
91   /**
92    * The closure to the callback @e cont
93    */
94   void *cont_cls;
95
96   /**
97    * The address of the id to be stored at
98    */
99   struct GNUNET_PeerIdentity *id;
100 };
101
102
103 /**
104  * Type of function used to differentiate between modified and not modified
105  * Sampler.
106  */
107 typedef void
108 (*RPS_get_peers_type) (void *cls);
109
110 /**
111  * Get one random peer out of the sampled peers.
112  *
113  * We might want to reinitialise this sampler after giving the
114  * corrsponding peer to the client.
115  * Only used internally
116  */
117 static void
118 sampler_get_rand_peer (void *cls);
119
120
121 /**
122  * Get one random peer out of the sampled peers.
123  *
124  * We might want to reinitialise this sampler after giving the
125  * corrsponding peer to the client.
126  */
127 static void
128 sampler_mod_get_rand_peer (void *cls);
129
130
131 /**
132  * Sampler with its own array of SamplerElements
133  */
134 struct RPS_Sampler
135 {
136   /**
137    * Number of sampler elements we hold.
138    */
139   unsigned int sampler_size;
140   //size_t size;
141
142   /**
143    * All sampler elements in one array.
144    */
145   struct RPS_SamplerElement **sampler_elements;
146
147   /**
148    * Maximum time a round takes
149    *
150    * Used in the context of RPS
151    */
152   struct GNUNET_TIME_Relative max_round_interval;
153
154   /**
155    * Stores the function to return peers. Which one it is depends on whether
156    * the Sampler is the modified one or not.
157    */
158   RPS_get_peers_type get_peers;
159
160   /**
161    * Head and tail for the DLL to store the #RPS_SamplerRequestHandle
162    */
163   struct RPS_SamplerRequestHandle *req_handle_head;
164   struct RPS_SamplerRequestHandle *req_handle_tail;
165
166   #ifdef TO_FILE
167   /**
168    * File name to log to
169    */
170   char *file_name;
171   #endif /* TO_FILE */
172 };
173
174 /**
175  * Closure to _get_n_rand_peers_ready_cb()
176  */
177 struct RPS_SamplerRequestHandle
178 {
179   /**
180    * DLL
181    */
182   struct RPS_SamplerRequestHandle *next;
183   struct RPS_SamplerRequestHandle *prev;
184
185   /**
186    * Number of peers we are waiting for.
187    */
188   uint32_t num_peers;
189
190   /**
191    * Number of peers we currently have.
192    */
193   uint32_t cur_num_peers;
194
195   /**
196    * Pointer to the array holding the ids.
197    */
198   struct GNUNET_PeerIdentity *ids;
199
200   /**
201    * Head and tail for the DLL to store the tasks for single requests
202    */
203   struct GetPeerCls *gpc_head;
204   struct GetPeerCls *gpc_tail;
205
206   /**
207    * Sampler.
208    */
209   struct RPS_Sampler *sampler;
210
211   /**
212    * Callback to be called when all ids are available.
213    */
214   RPS_sampler_n_rand_peers_ready_cb callback;
215
216   /**
217    * Closure given to the callback
218    */
219   void *cls;
220 };
221
222 ///**
223 // * Global sampler variable.
224 // */
225 //struct RPS_Sampler *sampler;
226
227
228 /**
229  * The minimal size for the extended sampler elements.
230  */
231 static size_t min_size;
232
233 /**
234  * The maximal size the extended sampler elements should grow to.
235  */
236 static size_t max_size;
237
238 /**
239  * The size the extended sampler elements currently have.
240  */
241 //static size_t extra_size;
242
243 /**
244  * Inedex to the sampler element that is the next to be returned
245  */
246 static uint32_t client_get_index;
247
248
249 /**
250  * Callback to _get_rand_peer() used by _get_n_rand_peers().
251  *
252  * Checks whether all n peers are available. If they are,
253  * give those back.
254  */
255 static void
256 check_n_peers_ready (void *cls,
257     const struct GNUNET_PeerIdentity *id)
258 {
259   struct RPS_SamplerRequestHandle *req_handle = cls;
260
261   req_handle->cur_num_peers++;
262   LOG (GNUNET_ERROR_TYPE_DEBUG,
263       "Got %" PRIX32 ". of %" PRIX32 " peers\n",
264       req_handle->cur_num_peers, req_handle->num_peers);
265
266   if (req_handle->num_peers == req_handle->cur_num_peers)
267   { /* All peers are ready -- return those to the client */
268     GNUNET_assert (NULL != req_handle->callback);
269
270     LOG (GNUNET_ERROR_TYPE_DEBUG,
271         "returning %" PRIX32 " peers to the client\n",
272         req_handle->num_peers);
273     req_handle->callback (req_handle->cls, req_handle->ids, req_handle->num_peers);
274
275     RPS_sampler_request_cancel (req_handle);
276   }
277 }
278
279
280 /**
281  * Get the size of the sampler.
282  *
283  * @param sampler the sampler to return the size of.
284  * @return the size of the sampler
285  */
286 unsigned int
287 RPS_sampler_get_size (struct RPS_Sampler *sampler)
288 {
289   return sampler->sampler_size;
290 }
291
292
293 /**
294  * Grow or shrink the size of the sampler.
295  *
296  * @param sampler the sampler to resize.
297  * @param new_size the new size of the sampler
298  */
299 static void
300 sampler_resize (struct RPS_Sampler *sampler, unsigned int new_size)
301 {
302   unsigned int old_size;
303   uint32_t i;
304
305   // TODO check min and max size
306
307   old_size = sampler->sampler_size;
308
309   if (old_size > new_size)
310   { /* Shrinking */
311
312     LOG (GNUNET_ERROR_TYPE_DEBUG,
313          "Shrinking sampler %d -> %d\n",
314          old_size,
315          new_size);
316
317     to_file (sampler->file_name,
318          "Shrinking sampler %d -> %d",
319          old_size,
320          new_size);
321
322     for (i = new_size ; i < old_size ; i++)
323     {
324       to_file (sampler->file_name,
325                "-%" PRIu32 ": %s",
326                i,
327                sampler->sampler_elements[i]->file_name);
328       RPS_sampler_elem_destroy (sampler->sampler_elements[i]);
329     }
330
331     GNUNET_array_grow (sampler->sampler_elements,
332                        sampler->sampler_size,
333                        new_size);
334     LOG (GNUNET_ERROR_TYPE_DEBUG,
335          "sampler->sampler_elements now points to %p\n",
336          sampler->sampler_elements);
337
338   }
339   else if (old_size < new_size)
340   { /* Growing */
341     LOG (GNUNET_ERROR_TYPE_DEBUG,
342          "Growing sampler %d -> %d\n",
343          old_size,
344          new_size);
345
346     to_file (sampler->file_name,
347          "Growing sampler %d -> %d",
348          old_size,
349          new_size);
350
351     GNUNET_array_grow (sampler->sampler_elements,
352         sampler->sampler_size,
353         new_size);
354
355     for (i = old_size ; i < new_size ; i++)
356     { /* Add new sampler elements */
357       sampler->sampler_elements[i] = RPS_sampler_elem_create ();
358
359       to_file (sampler->file_name,
360                "+%" PRIu32 ": %s",
361                i,
362                sampler->sampler_elements[i]->file_name);
363     }
364   }
365   else
366   {
367     LOG (GNUNET_ERROR_TYPE_DEBUG, "Size remains the same -- nothing to do\n");
368     return;
369   }
370
371   GNUNET_assert (sampler->sampler_size == new_size);
372 }
373
374
375 /**
376  * Grow or shrink the size of the sampler.
377  *
378  * @param sampler the sampler to resize.
379  * @param new_size the new size of the sampler
380  */
381 void
382 RPS_sampler_resize (struct RPS_Sampler *sampler, unsigned int new_size)
383 {
384   GNUNET_assert (0 < new_size);
385   sampler_resize (sampler, new_size);
386 }
387
388
389 /**
390  * Empty the sampler.
391  *
392  * @param sampler the sampler to empty.
393  * @param new_size the new size of the sampler
394  */
395 static void
396 sampler_empty (struct RPS_Sampler *sampler)
397 {
398   sampler_resize (sampler, 0);
399 }
400
401
402 /**
403  * Initialise a tuple of sampler elements.
404  *
405  * @param init_size the size the sampler is initialised with
406  * @param max_round_interval maximum time a round takes
407  * @return a handle to a sampler that consists of sampler elements.
408  */
409 struct RPS_Sampler *
410 RPS_sampler_init (size_t init_size,
411                   struct GNUNET_TIME_Relative max_round_interval)
412 {
413   struct RPS_Sampler *sampler;
414
415   /* Initialise context around extended sampler */
416   min_size = 10; // TODO make input to _samplers_init()
417   max_size = 1000; // TODO make input to _samplers_init()
418
419   sampler = GNUNET_new (struct RPS_Sampler);
420
421   #ifdef TO_FILE
422   sampler->file_name = create_file ("sampler-");
423
424   LOG (GNUNET_ERROR_TYPE_DEBUG,
425        "Initialised sampler %s\n",
426        sampler->file_name);
427   #endif /* TO_FILE */
428
429   sampler->max_round_interval = max_round_interval;
430   sampler->get_peers = sampler_get_rand_peer;
431   //sampler->sampler_elements = GNUNET_new_array(init_size, struct GNUNET_PeerIdentity);
432   //GNUNET_array_grow (sampler->sampler_elements, sampler->sampler_size, min_size);
433   RPS_sampler_resize (sampler, init_size);
434
435   client_get_index = 0;
436
437   //GNUNET_assert (init_size == sampler->sampler_size);
438   return sampler;
439 }
440
441 /**
442  * Initialise a modified tuple of sampler elements.
443  *
444  * @param init_size the size the sampler is initialised with
445  * @param max_round_interval maximum time a round takes
446  * @return a handle to a sampler that consists of sampler elements.
447  */
448 struct RPS_Sampler *
449 RPS_sampler_mod_init (size_t init_size,
450                       struct GNUNET_TIME_Relative max_round_interval)
451 {
452   struct RPS_Sampler *sampler;
453
454   sampler = RPS_sampler_init (init_size, max_round_interval);
455   sampler->get_peers = sampler_mod_get_rand_peer;
456
457   LOG (GNUNET_ERROR_TYPE_DEBUG,
458        "Initialised modified sampler %s\n",
459        sampler->file_name);
460   to_file (sampler->file_name,
461            "This is a modified sampler");
462
463   return sampler;
464 }
465
466
467 /**
468  * A fuction to update every sampler in the given list
469  *
470  * @param sampler the sampler to update.
471  * @param id the PeerID that is put in the sampler
472  */
473   void
474 RPS_sampler_update (struct RPS_Sampler *sampler,
475                     const struct GNUNET_PeerIdentity *id)
476 {
477   uint32_t i;
478
479   to_file (sampler->file_name,
480            "Got %s",
481            GNUNET_i2s_full (id));
482
483   for (i = 0 ; i < sampler->sampler_size ; i++)
484   {
485     RPS_sampler_elem_next (sampler->sampler_elements[i],
486                            id);
487   }
488
489 }
490
491
492 /**
493  * Reinitialise all previously initialised sampler elements with the given value.
494  *
495  * Used to get rid of a PeerID.
496  *
497  * @param sampler the sampler to reinitialise a sampler element in.
498  * @param id the id of the sampler elements to update.
499  */
500   void
501 RPS_sampler_reinitialise_by_value (struct RPS_Sampler *sampler,
502                                    const struct GNUNET_PeerIdentity *id)
503 {
504   uint32_t i;
505
506   for (i = 0; i < sampler->sampler_size; i++)
507   {
508     if (0 == GNUNET_CRYPTO_cmp_peer_identity(id,
509           &(sampler->sampler_elements[i]->peer_id)) )
510     {
511       LOG (GNUNET_ERROR_TYPE_DEBUG, "Reinitialising sampler\n");
512       to_file (sampler->sampler_elements[i]->file_name,
513                "--- non-active");
514       RPS_sampler_elem_reinit (sampler->sampler_elements[i]);
515     }
516   }
517 }
518
519
520 /**
521  * Get one random peer out of the sampled peers.
522  *
523  * We might want to reinitialise this sampler after giving the
524  * corrsponding peer to the client.
525  * Only used internally
526  */
527 static void
528 sampler_get_rand_peer (void *cls)
529 {
530   struct GetPeerCls *gpc = cls;
531   uint32_t r_index;
532   struct RPS_Sampler *sampler;
533
534   gpc->get_peer_task = NULL;
535   sampler = gpc->req_handle->sampler;
536
537   /**;
538    * Choose the r_index of the peer we want to return
539    * at random from the interval of the gossip list
540    */
541   r_index = GNUNET_CRYPTO_random_u64 (GNUNET_CRYPTO_QUALITY_STRONG,
542       sampler->sampler_size);
543
544   if (EMPTY == sampler->sampler_elements[r_index]->is_empty)
545   {
546     //LOG (GNUNET_ERROR_TYPE_DEBUG,
547     //     "Not returning randomly selected, empty PeerID. - Rescheduling.\n");
548
549     /* FIXME no active wait - get notified, when new id arrives?
550      * Might also be a freshly emptied one. Others might still contain ids.
551      * Counter?
552      */
553     gpc->get_peer_task =
554       GNUNET_SCHEDULER_add_delayed (
555           GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 1),
556           &sampler_get_rand_peer,
557           cls);
558     return;
559   }
560
561   GNUNET_CONTAINER_DLL_remove (gpc->req_handle->gpc_head,
562                                gpc->req_handle->gpc_tail,
563                                gpc);
564   *gpc->id = sampler->sampler_elements[r_index]->peer_id;
565   gpc->cont (gpc->cont_cls, gpc->id);
566
567   GNUNET_free (gpc);
568 }
569
570
571 /**
572  * Get one random peer out of the sampled peers.
573  *
574  * We might want to reinitialise this sampler after giving the
575  * corrsponding peer to the client.
576  */
577 static void
578 sampler_mod_get_rand_peer (void *cls)
579 {
580   struct GetPeerCls *gpc = cls;
581   struct RPS_SamplerElement *s_elem;
582   struct GNUNET_TIME_Relative last_request_diff;
583   struct RPS_Sampler *sampler;
584
585   gpc->get_peer_task = NULL;
586   sampler = gpc->req_handle->sampler;
587
588   LOG (GNUNET_ERROR_TYPE_DEBUG, "Single peer was requested\n");
589
590   /* Cycle the #client_get_index one step further */
591   client_get_index = (client_get_index + 1) % sampler->sampler_size;
592
593   s_elem = sampler->sampler_elements[client_get_index];
594   *gpc->id = s_elem->peer_id;
595   GNUNET_assert (NULL != s_elem);
596
597   if (EMPTY == s_elem->is_empty)
598   {
599     LOG (GNUNET_ERROR_TYPE_DEBUG,
600          "Sampler_mod element empty, rescheduling.\n");
601     GNUNET_assert (NULL == gpc->get_peer_task);
602     gpc->get_peer_task =
603       GNUNET_SCHEDULER_add_delayed (sampler->max_round_interval,
604                                     &sampler_mod_get_rand_peer,
605                                     cls);
606     return;
607   }
608
609   /* Check whether we may use this sampler to give it back to the client */
610   if (GNUNET_TIME_UNIT_FOREVER_ABS.abs_value_us != s_elem->last_client_request.abs_value_us)
611   {
612     last_request_diff =
613       GNUNET_TIME_absolute_get_difference (s_elem->last_client_request,
614                                            GNUNET_TIME_absolute_get ());
615     /* We're not going to give it back now if it was
616      * already requested by a client this round */
617     if (last_request_diff.rel_value_us < sampler->max_round_interval.rel_value_us)
618     {
619       LOG (GNUNET_ERROR_TYPE_DEBUG,
620           "Last client request on this sampler was less than max round interval ago -- scheduling for later\n");
621       ///* How many time remains untile the next round has started? */
622       //inv_last_request_diff =
623       //  GNUNET_TIME_absolute_get_difference (last_request_diff,
624       //                                       sampler->max_round_interval);
625       // add a little delay
626       /* Schedule it one round later */
627       GNUNET_assert (NULL == gpc->get_peer_task);
628       gpc->get_peer_task =
629         GNUNET_SCHEDULER_add_delayed (sampler->max_round_interval,
630                                       &sampler_mod_get_rand_peer,
631                                       cls);
632       return;
633     }
634     // TODO add other reasons to wait here
635   }
636
637   s_elem->last_client_request = GNUNET_TIME_absolute_get ();
638
639   GNUNET_CONTAINER_DLL_remove (gpc->req_handle->gpc_head,
640                                gpc->req_handle->gpc_tail,
641                                gpc);
642   gpc->cont (gpc->cont_cls, gpc->id);
643   GNUNET_free (gpc);
644 }
645
646
647 /**
648  * Get n random peers out of the sampled peers.
649  *
650  * We might want to reinitialise this sampler after giving the
651  * corrsponding peer to the client.
652  * Random with or without consumption?
653  *
654  * @param sampler the sampler to get peers from.
655  * @param cb callback that will be called once the ids are ready.
656  * @param cls closure given to @a cb
657  * @param for_client #GNUNET_YES if result is used for client,
658  *                   #GNUNET_NO if used internally
659  * @param num_peers the number of peers requested
660  */
661 struct RPS_SamplerRequestHandle *
662 RPS_sampler_get_n_rand_peers (struct RPS_Sampler *sampler,
663                               RPS_sampler_n_rand_peers_ready_cb cb,
664                               void *cls, uint32_t num_peers)
665 {
666   GNUNET_assert (0 != sampler->sampler_size);
667   if (0 == num_peers)
668     return NULL;
669
670   // TODO check if we have too much (distinct) sampled peers
671   uint32_t i;
672   struct RPS_SamplerRequestHandle *req_handle;
673   struct GetPeerCls *gpc;
674
675   req_handle = GNUNET_new (struct RPS_SamplerRequestHandle);
676   req_handle->num_peers = num_peers;
677   req_handle->cur_num_peers = 0;
678   req_handle->ids = GNUNET_new_array (num_peers, struct GNUNET_PeerIdentity);
679   req_handle->sampler = sampler;
680   req_handle->callback = cb;
681   req_handle->cls = cls;
682   GNUNET_CONTAINER_DLL_insert (sampler->req_handle_head,
683                                sampler->req_handle_tail,
684                                req_handle);
685
686   LOG (GNUNET_ERROR_TYPE_DEBUG,
687       "Scheduling requests for %" PRIu32 " peers\n", num_peers);
688
689   for (i = 0 ; i < num_peers ; i++)
690   {
691     gpc = GNUNET_new (struct GetPeerCls);
692     gpc->req_handle = req_handle;
693     gpc->cont = check_n_peers_ready;
694     gpc->cont_cls = req_handle;
695     gpc->id = &req_handle->ids[i];
696
697     GNUNET_CONTAINER_DLL_insert (req_handle->gpc_head,
698                                  req_handle->gpc_tail,
699                                  gpc);
700     // maybe add a little delay
701     gpc->get_peer_task = GNUNET_SCHEDULER_add_now (sampler->get_peers,
702                                                    gpc);
703   }
704   return req_handle;
705 }
706
707 /**
708  * Cancle a request issued through #RPS_sampler_n_rand_peers_ready_cb.
709  *
710  * @param req_handle the handle to the request
711  */
712 void
713 RPS_sampler_request_cancel (struct RPS_SamplerRequestHandle *req_handle)
714 {
715   struct GetPeerCls *i;
716
717   while (NULL != (i = req_handle->gpc_head) )
718   {
719     GNUNET_CONTAINER_DLL_remove (req_handle->gpc_head,
720                                  req_handle->gpc_tail,
721                                  i);
722     if (NULL != i->get_peer_task)
723     {
724       GNUNET_SCHEDULER_cancel (i->get_peer_task);
725     }
726     GNUNET_free (i);
727   }
728   GNUNET_CONTAINER_DLL_remove (req_handle->sampler->req_handle_head,
729                                req_handle->sampler->req_handle_tail,
730                                req_handle);
731   GNUNET_free (req_handle);
732 }
733
734
735 /**
736  * Counts how many Samplers currently hold a given PeerID.
737  *
738  * @param sampler the sampler to count ids in.
739  * @param id the PeerID to count.
740  *
741  * @return the number of occurrences of id.
742  */
743   uint32_t
744 RPS_sampler_count_id (struct RPS_Sampler *sampler,
745                       const struct GNUNET_PeerIdentity *id)
746 {
747   uint32_t count;
748   uint32_t i;
749
750   count = 0;
751   for ( i = 0 ; i < sampler->sampler_size ; i++ )
752   {
753     if ( 0 == GNUNET_CRYPTO_cmp_peer_identity (&sampler->sampler_elements[i]->peer_id, id)
754         && EMPTY != sampler->sampler_elements[i]->is_empty)
755       count++;
756   }
757   return count;
758 }
759
760
761 /**
762  * Cleans the sampler.
763  */
764   void
765 RPS_sampler_destroy (struct RPS_Sampler *sampler)
766 {
767   if (NULL != sampler->req_handle_head)
768   {
769     LOG (GNUNET_ERROR_TYPE_WARNING,
770         "There are still pending requests. Going to remove them.\n");
771     while (NULL != sampler->req_handle_head)
772     {
773       RPS_sampler_request_cancel (sampler->req_handle_head);
774     }
775   }
776   sampler_empty (sampler);
777   GNUNET_free (sampler);
778 }
779
780 /* end of gnunet-service-rps.c */