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    * @brief The Callback to call on updates
  74    */
  75   SamplerNotifyUpdateCB notify_cb;
  76
  77   /**
  78    * @brief The according closure.
  79    */
  80   void *cls;
  81
  82   /**
  83    * @brief Next element in DLL.
  84    */
  85   struct SamplerNotifyUpdateCTX *next;
  86
  87   /**
  88    * @brief Previous element in DLL.
  89    */
  90   struct SamplerNotifyUpdateCTX *prev;
  91 };
  92
  93
  94 /**
  95  * Type of function used to differentiate between modified and not modified
  96  * Sampler.
  97  */
  98 typedef void
  99 (*RPS_get_peers_type) (void *cls);
 100
 101
 102 /**
 103  * Get one random peer out of the sampled peers.
 104  *
 105  * We might want to reinitialise this sampler after giving the
 106  * corrsponding peer to the client.
 107  */
 108 static void
 109 sampler_mod_get_rand_peer(void *cls);
 110
 111
 112 /**
 113  * Closure to _get_n_rand_peers_ready_cb()
 114  */
 115 struct RPS_SamplerRequestHandle {
 116   /**
 117    * DLL
 118    */
 119   struct RPS_SamplerRequestHandle *next;
 120   struct RPS_SamplerRequestHandle *prev;
 121
 122   /**
 123    * Number of peers we are waiting for.
 124    */
 125   uint32_t num_peers;
 126
 127   /**
 128    * Number of peers we currently have.
 129    */
 130   uint32_t cur_num_peers;
 131
 132   /**
 133    * Pointer to the array holding the ids.
 134    */
 135   struct GNUNET_PeerIdentity *ids;
 136
 137   /**
 138    * Head and tail for the DLL to store the tasks for single requests
 139    */
 140   struct GetPeerCls *gpc_head;
 141   struct GetPeerCls *gpc_tail;
 142
 143   /**
 144    * Sampler.
 145    */
 146   struct RPS_Sampler *sampler;
 147
 148   /**
 149    * Callback to be called when all ids are available.
 150    */
 151   RPS_sampler_n_rand_peers_ready_cb callback;
 152
 153   /**
 154    * Closure given to the callback
 155    */
 156   void *cls;
 157 };
 158
 159
 160 /**
 161  * Closure to _get_rand_peer_info()
 162  */
 163 struct RPS_SamplerRequestHandleSingleInfo {
 164   /**
 165    * DLL
 166    */
 167   struct RPS_SamplerRequestHandleSingleInfo *next;
 168   struct RPS_SamplerRequestHandleSingleInfo *prev;
 169
 170   /**
 171    * Pointer to the id
 172    */
 173   struct GNUNET_PeerIdentity *id;
 174
 175   /**
 176    * Head and tail for the DLL to store the tasks for single requests
 177    */
 178   struct GetPeerCls *gpc_head;
 179   struct GetPeerCls *gpc_tail;
 180
 181   /**
 182    * Sampler.
 183    */
 184   struct RPS_Sampler *sampler;
 185
 186   /**
 187    * Callback to be called when all ids are available.
 188    */
 189   RPS_sampler_sinlge_info_ready_cb callback;
 190
 191   /**
 192    * Closure given to the callback
 193    */
 194   void *cls;
 195 };
 196
 197
 198 ///**
 199 // * Global sampler variable.
 200 // */
 201 //struct RPS_Sampler *sampler;
 202
 203
 204 /**
 205  * The minimal size for the extended sampler elements.
 206  */
 207 static size_t min_size;
 208
 209 /**
 210  * The maximal size the extended sampler elements should grow to.
 211  */
 212 static size_t max_size;
 213
 214 /**
 215  * The size the extended sampler elements currently have.
 216  */
 217 //static size_t extra_size;
 218
 219 /**
 220  * Inedex to the sampler element that is the next to be returned
 221  */
 222 static uint32_t client_get_index;
 223
 224
 225 /**
 226  * Initialise a modified tuple of sampler elements.
 227  *
 228  * @param init_size the size the sampler is initialised with
 229  * @param max_round_interval maximum time a round takes
 230  * @return a handle to a sampler that consists of sampler elements.
 231  */
 232 struct RPS_Sampler *
 233 RPS_sampler_mod_init(size_t init_size,
 234                      struct GNUNET_TIME_Relative max_round_interval)
 235 {
 236   struct RPS_Sampler *sampler;
 237
 238   /* Initialise context around extended sampler */
 239   min_size = 10; // TODO make input to _samplers_init()
 240   max_size = 1000; // TODO make input to _samplers_init()
 241
 242   sampler = GNUNET_new(struct RPS_Sampler);
 243   sampler->max_round_interval = max_round_interval;
 244   sampler->get_peers = sampler_mod_get_rand_peer;
 245   //sampler->sampler_elements = GNUNET_new_array(init_size, struct GNUNET_PeerIdentity);
 246   //GNUNET_array_grow (sampler->sampler_elements, sampler->sampler_size, min_size);
 247
 248   client_get_index = 0;
 249
 250   //GNUNET_assert (init_size == sampler->sampler_size);
 251
 252   RPS_sampler_resize(sampler, init_size);
 253
 254   return sampler;
 255 }
 256
 257
 258 /**
 259  * @brief Compute the probability that we already observed all peers from a
 260  * biased stream of peer ids.
 261  *
 262  * Deficiency factor:
 263  * As introduced by Brahms: Factor between the number of unique ids in a
 264  * truly random stream and number of unique ids in the gossip stream.
 265  *
 266  * @param num_peers_estim The estimated number of peers in the network
 267  * @param num_peers_observed The number of peers the given element has observed
 268  * @param deficiency_factor A factor that catches the 'bias' of a random stream
 269  * of peer ids
 270  *
 271  * @return The estimated probability
 272  */
 273 static double
 274 prob_observed_n_peers(uint32_t num_peers_estim,
 275                       uint32_t num_peers_observed,
 276                       double deficiency_factor)
 277 {
 278   uint32_t num_peers = num_peers_estim * (1 / deficiency_factor);
 279   uint64_t sum = 0;
 280
 281   for (uint32_t i = 0; i < num_peers; i++)
 282     {
 283       uint64_t a = pow(-1, num_peers - i);
 284       uint64_t b = binom(num_peers, i);
 285       uint64_t c = pow(i, num_peers_observed);
 286       sum += a * b * c;
 287     }
 288
 289   return sum / (double)pow(num_peers, num_peers_observed);
 290 }
 291
 292
 293 /**
 294  * Get one random peer out of the sampled peers.
 295  *
 296  * This reinitialises the queried sampler element.
 297  */
 298 static void
 299 sampler_mod_get_rand_peer(void *cls)
 300 {
 301   struct GetPeerCls *gpc = cls;
 302   struct RPS_SamplerElement *s_elem;
 303   struct GNUNET_TIME_Relative last_request_diff;
 304   struct RPS_Sampler *sampler;
 305   double prob_observed_n;
 306   uint32_t num_observed;
 307
 308   gpc->get_peer_task = NULL;
 309   gpc->notify_ctx = NULL;
 310   GNUNET_assert((NULL != gpc->req_handle) ||
 311                 (NULL != gpc->req_single_info_handle));
 312   if (NULL != gpc->req_handle)
 313     sampler = gpc->req_handle->sampler;
 314   else
 315     sampler = gpc->req_single_info_handle->sampler;
 316
 317   LOG(GNUNET_ERROR_TYPE_DEBUG, "Single peer was requested\n");
 318
 319   /* Cycle the #client_get_index one step further */
 320   client_get_index = (client_get_index + 1) % sampler->sampler_size;
 321
 322   s_elem = sampler->sampler_elements[client_get_index];
 323   *gpc->id = s_elem->peer_id;
 324   GNUNET_assert(NULL != s_elem);
 325
 326   if (EMPTY == s_elem->is_empty)
 327     {
 328       LOG(GNUNET_ERROR_TYPE_DEBUG,
 329           "Sampler_mod element empty, rescheduling.\n");
 330       GNUNET_assert(NULL == gpc->notify_ctx);
 331       gpc->notify_ctx =
 332         sampler_notify_on_update(sampler,
 333                                  &sampler_mod_get_rand_peer,
 334                                  gpc);
 335       return;
 336     }
 337
 338   /* Check whether we may use this sampler to give it back to the client */
 339   if (GNUNET_TIME_UNIT_FOREVER_ABS.abs_value_us != s_elem->last_client_request.abs_value_us)
 340     {
 341       // TODO remove this condition at least for the client sampler
 342       last_request_diff =
 343         GNUNET_TIME_absolute_get_difference(s_elem->last_client_request,
 344                                             GNUNET_TIME_absolute_get());
 345       /* We're not going to give it back now if it was
 346        * already requested by a client this round */
 347       if (last_request_diff.rel_value_us < sampler->max_round_interval.rel_value_us)
 348         {
 349           LOG(GNUNET_ERROR_TYPE_DEBUG,
 350               "Last client request on this sampler was less than max round interval ago -- scheduling for later\n");
 351           ///* How many time remains untile the next round has started? */
 352           //inv_last_request_diff =
 353           //  GNUNET_TIME_absolute_get_difference (last_request_diff,
 354           //                                       sampler->max_round_interval);
 355           // add a little delay
 356           /* Schedule it one round later */
 357           GNUNET_assert(NULL == gpc->notify_ctx);
 358           gpc->notify_ctx =
 359             sampler_notify_on_update(sampler,
 360                                      &sampler_mod_get_rand_peer,
 361                                      gpc);
 362           return;
 363         }
 364     }
 365   if (2 > s_elem->num_peers)
 366     {
 367       LOG(GNUNET_ERROR_TYPE_DEBUG,
 368           "This s_elem saw less than two peers -- scheduling for later\n");
 369       GNUNET_assert(NULL == gpc->notify_ctx);
 370       gpc->notify_ctx =
 371         sampler_notify_on_update(sampler,
 372                                  &sampler_mod_get_rand_peer,
 373                                  gpc);
 374       return;
 375     }
 376   /* compute probability */
 377   /* Currently disabled due to numerical limitations */
 378   //prob_observed_n = prob_observed_n_peers (sampler->num_peers_estim,
 379   //                                         s_elem->num_peers,
 380   //                                         sampler->deficiency_factor);
 381   //LOG (GNUNET_ERROR_TYPE_DEBUG,
 382   //    "Computed sample - prob %f, %" PRIu32 " peers, n: %" PRIu32 ", roh: %f\n",
 383   //    prob_observed_n,
 384   //    s_elem->num_peers,
 385   //    sampler->num_peers_estim,
 386   //    sampler->deficiency_factor);
 387   ///* check if probability is above desired */
 388   //if (prob_observed_n < sampler->desired_probability)
 389   //{
 390   //  LOG (GNUNET_ERROR_TYPE_DEBUG,
 391   //      "Probability of having observed all peers (%f) too small ( < %f).\n",
 392   //      prob_observed_n,
 393   //      sampler->desired_probability);
 394   //  GNUNET_assert (NULL == gpc->notify_ctx);
 395   //  gpc->notify_ctx =
 396   //    sampler_notify_on_update (sampler,
 397   //                              &sampler_mod_get_rand_peer,
 398   //                              gpc);
 399   //  return;
 400   //}
 401   /* More reasons to wait could be added here */
 402
 403 //  GNUNET_STATISTICS_set (stats,
 404 //                         "# client sampler element input",
 405 //                         s_elem->num_peers,
 406 //                         GNUNET_NO);
 407 //  GNUNET_STATISTICS_set (stats,
 408 //                         "# client sampler element change",
 409 //                         s_elem->num_change,
 410 //                         GNUNET_NO);
 411
 412   num_observed = s_elem->num_peers;
 413   RPS_sampler_elem_reinit(s_elem);
 414   s_elem->last_client_request = GNUNET_TIME_absolute_get();
 415
 416   if (NULL != gpc->req_handle)
 417     {
 418       GNUNET_CONTAINER_DLL_remove(gpc->req_handle->gpc_head,
 419                                   gpc->req_handle->gpc_tail,
 420                                   gpc);
 421     }
 422   else
 423     {
 424       GNUNET_CONTAINER_DLL_remove(gpc->req_single_info_handle->gpc_head,
 425                                   gpc->req_single_info_handle->gpc_tail,
 426                                   gpc);
 427     }
 428   gpc->cont(gpc->cont_cls, gpc->id, prob_observed_n, num_observed);
 429   GNUNET_free(gpc);
 430 }
 431
 432
 433 /* end of gnunet-service-rps.c */
 434