src/rps/rps-sampler_client.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      SPDX-License-Identifier: AGPL3.0-or-later
  19 */
  20
  21 /**
  22  * @file rps/gnunet-service-rps_sampler.c
  23  * @brief sampler implementation
  24  * @author Julius Bünger
  25  */
  26 #include "platform.h"
  27 #include "gnunet_util_lib.h"
  28 #include "gnunet_statistics_service.h"
  29 #include "rps.h"
  30
  31 #include "rps-sampler_common.h"
  32 #include "gnunet-service-rps_sampler.h"
  33 #include "gnunet-service-rps_sampler_elem.h"
  34
  35 #include <math.h>
  36 #include <inttypes.h>
  37
  38 #include "rps-test_util.h"
  39
  40 #define LOG(kind, ...) GNUNET_log_from(kind,"rps-sampler",__VA_ARGS__)
  41
  42
  43 // multiple 'clients'?
  44
  45 // TODO check for overflows
  46
  47 // TODO align message structs
  48
  49 // hist_size_init, hist_size_max
  50
  51 /***********************************************************************
  52  * WARNING: This section needs to be reviewed regarding the use of
  53  * functions providing (pseudo)randomness!
  54 ***********************************************************************/
  55
  56 // TODO care about invalid input of the caller (size 0 or less...)
  57
  58 /**
  59  * @brief Callback called each time a new peer was put into the sampler
  60  *
  61  * @param cls A possibly given closure
  62  */
  63 typedef void
  64 (*SamplerNotifyUpdateCB) (void *cls);
  65
  66 /**
  67  * @brief Context for a callback. Contains callback and closure.
  68  *
  69  * Meant to be an entry in an DLL.
  70  */
  71 struct SamplerNotifyUpdateCTX
  72 {
  73   /**
  74    * @brief The Callback to call on updates
  75    */
  76   SamplerNotifyUpdateCB notify_cb;
  77
  78   /**
  79    * @brief The according closure.
  80    */
  81   void *cls;
  82
  83   /**
  84    * @brief Next element in DLL.
  85    */
  86   struct SamplerNotifyUpdateCTX *next;
  87
  88   /**
  89    * @brief Previous element in DLL.
  90    */
  91   struct SamplerNotifyUpdateCTX *prev;
  92 };
  93
  94
  95 /**
  96  * Type of function used to differentiate between modified and not modified
  97  * Sampler.
  98  */
  99 typedef void
 100 (*RPS_get_peers_type) (void *cls);
 101
 102
 103 /**
 104  * Get one random peer out of the sampled peers.
 105  *
 106  * We might want to reinitialise this sampler after giving the
 107  * corrsponding peer to the client.
 108  */
 109 static void
 110 sampler_mod_get_rand_peer (void *cls);
 111
 112
 113 /**
 114  * Closure to _get_n_rand_peers_ready_cb()
 115  */
 116 struct RPS_SamplerRequestHandle
 117 {
 118   /**
 119    * DLL
 120    */
 121   struct RPS_SamplerRequestHandle *next;
 122   struct RPS_SamplerRequestHandle *prev;
 123
 124   /**
 125    * Number of peers we are waiting for.
 126    */
 127   uint32_t num_peers;
 128
 129   /**
 130    * Number of peers we currently have.
 131    */
 132   uint32_t cur_num_peers;
 133
 134   /**
 135    * Pointer to the array holding the ids.
 136    */
 137   struct GNUNET_PeerIdentity *ids;
 138
 139   /**
 140    * Head and tail for the DLL to store the tasks for single requests
 141    */
 142   struct GetPeerCls *gpc_head;
 143   struct GetPeerCls *gpc_tail;
 144
 145   /**
 146    * Sampler.
 147    */
 148   struct RPS_Sampler *sampler;
 149
 150   /**
 151    * Callback to be called when all ids are available.
 152    */
 153   RPS_sampler_n_rand_peers_ready_cb callback;
 154
 155   /**
 156    * Closure given to the callback
 157    */
 158   void *cls;
 159 };
 160
 161
 162 /**
 163  * Closure to _get_rand_peer_info()
 164  */
 165 struct RPS_SamplerRequestHandleSingleInfo
 166 {
 167   /**
 168    * DLL
 169    */
 170   struct RPS_SamplerRequestHandleSingleInfo *next;
 171   struct RPS_SamplerRequestHandleSingleInfo *prev;
 172
 173   /**
 174    * Pointer to the id
 175    */
 176   struct GNUNET_PeerIdentity *id;
 177
 178   /**
 179    * Head and tail for the DLL to store the tasks for single requests
 180    */
 181   struct GetPeerCls *gpc_head;
 182   struct GetPeerCls *gpc_tail;
 183
 184   /**
 185    * Sampler.
 186    */
 187   struct RPS_Sampler *sampler;
 188
 189   /**
 190    * Callback to be called when all ids are available.
 191    */
 192   RPS_sampler_sinlge_info_ready_cb callback;
 193
 194   /**
 195    * Closure given to the callback
 196    */
 197   void *cls;
 198 };
 199
 200
 201 ///**
 202 // * Global sampler variable.
 203 // */
 204 //struct RPS_Sampler *sampler;
 205
 206
 207 /**
 208  * The minimal size for the extended sampler elements.
 209  */
 210 static size_t min_size;
 211
 212 /**
 213  * The maximal size the extended sampler elements should grow to.
 214  */
 215 static size_t max_size;
 216
 217 /**
 218  * The size the extended sampler elements currently have.
 219  */
 220 //static size_t extra_size;
 221
 222 /**
 223  * Inedex to the sampler element that is the next to be returned
 224  */
 225 static uint32_t client_get_index;
 226
 227
 228 /**
 229  * Initialise a modified tuple of sampler elements.
 230  *
 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.
 234  */
 235 struct RPS_Sampler *
 236 RPS_sampler_mod_init (size_t init_size,
 237                       struct GNUNET_TIME_Relative max_round_interval)
 238 {
 239   struct RPS_Sampler *sampler;
 240
 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()
 244
 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);
 250
 251   client_get_index = 0;
 252
 253   //GNUNET_assert (init_size == sampler->sampler_size);
 254
 255   RPS_sampler_resize (sampler, init_size);
 256
 257   return sampler;
 258 }
 259
 260
 261 /**
 262  * @brief Compute the probability that we already observed all peers from a
 263  * biased stream of peer ids.
 264  *
 265  * Deficiency factor:
 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.
 268  *
 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
 272  * of peer ids
 273  *
 274  * @return The estimated probability
 275  */
 276 static double
 277 prob_observed_n_peers (uint32_t num_peers_estim,
 278                        uint32_t num_peers_observed,
 279                        double deficiency_factor)
 280 {
 281   uint32_t num_peers = num_peers_estim * (1/deficiency_factor);
 282   uint64_t sum = 0;
 283
 284   for (uint32_t i = 0; i < num_peers; i++)
 285   {
 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);
 289     sum += a * b * c;
 290   }
 291
 292   return sum / (double) pow (num_peers, num_peers_observed);
 293 }
 294
 295
 296 /**
 297  * Get one random peer out of the sampled peers.
 298  *
 299  * This reinitialises the queried sampler element.
 300  */
 301 static void
 302 sampler_mod_get_rand_peer (void *cls)
 303 {
 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;
 310
 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;
 317   else
 318     sampler = gpc->req_single_info_handle->sampler;
 319
 320   LOG (GNUNET_ERROR_TYPE_DEBUG, "Single peer was requested\n");
 321
 322   /* Cycle the #client_get_index one step further */
 323   client_get_index = (client_get_index + 1) % sampler->sampler_size;
 324
 325   s_elem = sampler->sampler_elements[client_get_index];
 326   *gpc->id = s_elem->peer_id;
 327   GNUNET_assert (NULL != s_elem);
 328
 329   if (EMPTY == s_elem->is_empty)
 330   {
 331     LOG (GNUNET_ERROR_TYPE_DEBUG,
 332          "Sampler_mod element empty, rescheduling.\n");
 333     GNUNET_assert (NULL == gpc->notify_ctx);
 334     gpc->notify_ctx =
 335       sampler_notify_on_update (sampler,
 336                                 &sampler_mod_get_rand_peer,
 337                                 gpc);
 338     return;
 339   }
 340
 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 != s_elem->last_client_request.abs_value_us)
 343   {
 344     // TODO remove this condition at least for the client sampler
 345     last_request_diff =
 346       GNUNET_TIME_absolute_get_difference (s_elem->last_client_request,
 347                                            GNUNET_TIME_absolute_get ());
 348     /* We're not going to give it back now if it was
 349      * already requested by a client this round */
 350     if (last_request_diff.rel_value_us < sampler->max_round_interval.rel_value_us)
 351     {
 352       LOG (GNUNET_ERROR_TYPE_DEBUG,
 353           "Last client request on this sampler was less than max round interval ago -- scheduling for later\n");
 354       ///* How many time remains untile the next round has started? */
 355       //inv_last_request_diff =
 356       //  GNUNET_TIME_absolute_get_difference (last_request_diff,
 357       //                                       sampler->max_round_interval);
 358       // add a little delay
 359       /* Schedule it one round later */
 360       GNUNET_assert (NULL == gpc->notify_ctx);
 361       gpc->notify_ctx =
 362         sampler_notify_on_update (sampler,
 363                                   &sampler_mod_get_rand_peer,
 364                                   gpc);
 365       return;
 366     }
 367   }
 368   if (2 > s_elem->num_peers)
 369   {
 370     LOG (GNUNET_ERROR_TYPE_DEBUG,
 371         "This s_elem saw less than two peers -- scheduling for later\n");
 372     GNUNET_assert (NULL == gpc->notify_ctx);
 373     gpc->notify_ctx =
 374       sampler_notify_on_update (sampler,
 375                                 &sampler_mod_get_rand_peer,
 376                                 gpc);
 377     return;
 378   }
 379   /* compute probability */
 380   /* Currently disabled due to numerical limitations */
 381   //prob_observed_n = prob_observed_n_peers (sampler->num_peers_estim,
 382   //                                         s_elem->num_peers,
 383   //                                         sampler->deficiency_factor);
 384   //LOG (GNUNET_ERROR_TYPE_DEBUG,
 385   //    "Computed sample - prob %f, %" PRIu32 " peers, n: %" PRIu32 ", roh: %f\n",
 386   //    prob_observed_n,
 387   //    s_elem->num_peers,
 388   //    sampler->num_peers_estim,
 389   //    sampler->deficiency_factor);
 390   ///* check if probability is above desired */
 391   //if (prob_observed_n < sampler->desired_probability)
 392   //{
 393   //  LOG (GNUNET_ERROR_TYPE_DEBUG,
 394   //      "Probability of having observed all peers (%f) too small ( < %f).\n",
 395   //      prob_observed_n,
 396   //      sampler->desired_probability);
 397   //  GNUNET_assert (NULL == gpc->notify_ctx);
 398   //  gpc->notify_ctx =
 399   //    sampler_notify_on_update (sampler,
 400   //                              &sampler_mod_get_rand_peer,
 401   //                              gpc);
 402   //  return;
 403   //}
 404   /* More reasons to wait could be added here */
 405
 406 //  GNUNET_STATISTICS_set (stats,
 407 //                         "# client sampler element input",
 408 //                         s_elem->num_peers,
 409 //                         GNUNET_NO);
 410 //  GNUNET_STATISTICS_set (stats,
 411 //                         "# client sampler element change",
 412 //                         s_elem->num_change,
 413 //                         GNUNET_NO);
 414
 415   num_observed = s_elem->num_peers;
 416   RPS_sampler_elem_reinit (s_elem);
 417   s_elem->last_client_request = GNUNET_TIME_absolute_get ();
 418
 419   if (NULL != gpc->req_handle)
 420   {
 421     GNUNET_CONTAINER_DLL_remove (gpc->req_handle->gpc_head,
 422                                  gpc->req_handle->gpc_tail,
 423                                  gpc);
 424   }
 425   else
 426   {
 427     GNUNET_CONTAINER_DLL_remove (gpc->req_single_info_handle->gpc_head,
 428                                  gpc->req_single_info_handle->gpc_tail,
 429                                  gpc);
 430   }
 431   gpc->cont (gpc->cont_cls, gpc->id, prob_observed_n, num_observed);
 432   GNUNET_free (gpc);
 433 }
 434
 435
 436 /* end of gnunet-service-rps.c */
 437