src/rps/rps-sampler_common.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/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  */
  45 struct 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  */
  72 struct 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  * @brief Add a callback that will be called when the next peer is inserted
 120  * into the sampler
 121  *
 122  * @param sampler The sampler on which update it will be called
 123  * @param notify_cb The callback
 124  * @param cls Closure given to the callback
 125  *
 126  * @return The context containing callback and closure
 127  */
 128 struct SamplerNotifyUpdateCTX *
 129 sampler_notify_on_update (struct RPS_Sampler *sampler,
 130                           SamplerNotifyUpdateCB notify_cb,
 131                           void *cls)
 132 {
 133   struct SamplerNotifyUpdateCTX *notify_ctx;
 134
 135   LOG (GNUNET_ERROR_TYPE_DEBUG,
 136       "Inserting new context for notification\n");
 137   notify_ctx = GNUNET_new (struct SamplerNotifyUpdateCTX);
 138   notify_ctx->notify_cb = notify_cb;
 139   notify_ctx->cls = cls;
 140   GNUNET_CONTAINER_DLL_insert (sampler->notify_ctx_head,
 141                                sampler->notify_ctx_tail,
 142                                notify_ctx);
 143   return notify_ctx;
 144 }
 145
 146
 147 /**
 148  * Get the size of the sampler.
 149  *
 150  * @param sampler the sampler to return the size of.
 151  * @return the size of the sampler
 152  */
 153 unsigned int
 154 RPS_sampler_get_size (struct RPS_Sampler *sampler)
 155 {
 156   return sampler->sampler_size;
 157 }
 158
 159
 160 /**
 161  * @brief Notify about update of the sampler.
 162  *
 163  * Call the callbacks that are waiting for notification on updates to the
 164  * sampler.
 165  *
 166  * @param sampler The sampler the updates are waiting for
 167  */
 168 static void
 169 notify_update (struct RPS_Sampler *sampler)
 170 {
 171   struct SamplerNotifyUpdateCTX *tmp_notify_head;
 172   struct SamplerNotifyUpdateCTX *tmp_notify_tail;
 173
 174   LOG (GNUNET_ERROR_TYPE_DEBUG,
 175       "Calling callbacks waiting for update notification.\n");
 176   tmp_notify_head = sampler->notify_ctx_head;
 177   tmp_notify_tail = sampler->notify_ctx_tail;
 178   sampler->notify_ctx_head = NULL;
 179   sampler->notify_ctx_tail = NULL;
 180   for (struct SamplerNotifyUpdateCTX *notify_iter = tmp_notify_head;
 181        NULL != tmp_notify_head;
 182        notify_iter = tmp_notify_head)
 183   {
 184     GNUNET_assert (NULL != notify_iter->notify_cb);
 185     GNUNET_CONTAINER_DLL_remove (tmp_notify_head,
 186                                  tmp_notify_tail,
 187                                  notify_iter);
 188     notify_iter->notify_cb (notify_iter->cls);
 189     GNUNET_free (notify_iter);
 190   }
 191 }
 192
 193
 194 /**
 195  * Update every sampler element of this sampler with given peer
 196  *
 197  * @param sampler the sampler to update.
 198  * @param id the PeerID that is put in the sampler
 199  */
 200   void
 201 RPS_sampler_update (struct RPS_Sampler *sampler,
 202                     const struct GNUNET_PeerIdentity *id)
 203 {
 204   for (uint32_t i = 0; i < sampler->sampler_size; i++)
 205   {
 206     RPS_sampler_elem_next (sampler->sampler_elements[i],
 207                            id);
 208   }
 209   notify_update (sampler);
 210 }
 211
 212
 213 /**
 214  * Reinitialise all previously initialised sampler elements with the given value.
 215  *
 216  * Used to get rid of a PeerID.
 217  *
 218  * @param sampler the sampler to reinitialise a sampler element in.
 219  * @param id the id of the sampler elements to update.
 220  */
 221   void
 222 RPS_sampler_reinitialise_by_value (struct RPS_Sampler *sampler,
 223                                    const struct GNUNET_PeerIdentity *id)
 224 {
 225   uint32_t i;
 226
 227   for (i = 0; i < sampler->sampler_size; i++)
 228   {
 229     if (0 == GNUNET_CRYPTO_cmp_peer_identity(id,
 230           &(sampler->sampler_elements[i]->peer_id)) )
 231     {
 232       LOG (GNUNET_ERROR_TYPE_DEBUG, "Reinitialising sampler\n");
 233       RPS_sampler_elem_reinit (sampler->sampler_elements[i]);
 234     }
 235   }
 236 }
 237
 238
 239 /**
 240  * Counts how many Samplers currently hold a given PeerID.
 241  *
 242  * @param sampler the sampler to count ids in.
 243  * @param id the PeerID to count.
 244  *
 245  * @return the number of occurrences of id.
 246  */
 247   uint32_t
 248 RPS_sampler_count_id (struct RPS_Sampler *sampler,
 249                       const struct GNUNET_PeerIdentity *id)
 250 {
 251   uint32_t count;
 252   uint32_t i;
 253
 254   count = 0;
 255   for ( i = 0 ; i < sampler->sampler_size ; i++ )
 256   {
 257     if ( 0 == GNUNET_CRYPTO_cmp_peer_identity (&sampler->sampler_elements[i]->peer_id, id)
 258         && EMPTY != sampler->sampler_elements[i]->is_empty)
 259       count++;
 260   }
 261   return count;
 262 }
 263
 264
 265 /**
 266  * Grow or shrink the size of the sampler.
 267  *
 268  * @param sampler the sampler to resize.
 269  * @param new_size the new size of the sampler
 270  */
 271 static void
 272 sampler_resize (struct RPS_Sampler *sampler, unsigned int new_size)
 273 {
 274   unsigned int old_size;
 275   uint32_t i;
 276
 277   // TODO check min and max size
 278
 279   old_size = sampler->sampler_size;
 280
 281   if (old_size > new_size)
 282   { /* Shrinking */
 283
 284     LOG (GNUNET_ERROR_TYPE_DEBUG,
 285          "Shrinking sampler %d -> %d\n",
 286          old_size,
 287          new_size);
 288
 289     for (i = new_size ; i < old_size ; i++)
 290     {
 291       RPS_sampler_elem_destroy (sampler->sampler_elements[i]);
 292     }
 293
 294     GNUNET_array_grow (sampler->sampler_elements,
 295                        sampler->sampler_size,
 296                        new_size);
 297     LOG (GNUNET_ERROR_TYPE_DEBUG,
 298          "sampler->sampler_elements now points to %p\n",
 299          sampler->sampler_elements);
 300
 301   }
 302   else if (old_size < new_size)
 303   { /* Growing */
 304     LOG (GNUNET_ERROR_TYPE_DEBUG,
 305          "Growing sampler %d -> %d\n",
 306          old_size,
 307          new_size);
 308
 309     GNUNET_array_grow (sampler->sampler_elements,
 310         sampler->sampler_size,
 311         new_size);
 312
 313     for (i = old_size ; i < new_size ; i++)
 314     { /* Add new sampler elements */
 315       sampler->sampler_elements[i] = RPS_sampler_elem_create ();
 316     }
 317   }
 318   else
 319   {
 320     LOG (GNUNET_ERROR_TYPE_DEBUG, "Size remains the same -- nothing to do\n");
 321     return;
 322   }
 323
 324   GNUNET_assert (sampler->sampler_size == new_size);
 325 }
 326
 327
 328 /**
 329  * Grow or shrink the size of the sampler.
 330  *
 331  * @param sampler the sampler to resize.
 332  * @param new_size the new size of the sampler
 333  */
 334 void
 335 RPS_sampler_resize (struct RPS_Sampler *sampler, unsigned int new_size)
 336 {
 337   GNUNET_assert (0 < new_size);
 338   sampler_resize (sampler, new_size);
 339 }
 340
 341
 342 /**
 343  * Empty the sampler.
 344  *
 345  * @param sampler the sampler to empty.
 346  * @param new_size the new size of the sampler
 347  */
 348 static void
 349 sampler_empty (struct RPS_Sampler *sampler)
 350 {
 351   sampler_resize (sampler, 0);
 352 }
 353
 354
 355 /**
 356  * Callback to _get_rand_peer() used by _get_n_rand_peers().
 357  *
 358  * Checks whether all n peers are available. If they are,
 359  * give those back.
 360  */
 361 static void
 362 check_n_peers_ready (void *cls,
 363                      const struct GNUNET_PeerIdentity *id)
 364 {
 365   struct RPS_SamplerRequestHandle *req_handle = cls;
 366   (void) id;
 367   RPS_sampler_n_rand_peers_ready_cb tmp_cb;
 368   struct GNUNET_PeerIdentity *peers;
 369   uint32_t num_peers;
 370   void *cb_cls;
 371
 372   req_handle->cur_num_peers++;
 373   LOG (GNUNET_ERROR_TYPE_DEBUG,
 374       "Got %" PRIX32 ". of %" PRIX32 " peers\n",
 375       req_handle->cur_num_peers, req_handle->num_peers);
 376
 377   if (req_handle->num_peers == req_handle->cur_num_peers)
 378   { /* All peers are ready -- return those to the client */
 379     GNUNET_assert (NULL != req_handle->callback);
 380
 381     LOG (GNUNET_ERROR_TYPE_DEBUG,
 382         "returning %" PRIX32 " peers to the client\n",
 383         req_handle->num_peers);
 384
 385     /* Copy pointers and peers temporarily as they
 386      * might be deleted from within the callback */
 387     tmp_cb = req_handle->callback;
 388     num_peers = req_handle->num_peers;
 389     peers = GNUNET_new_array (num_peers, struct GNUNET_PeerIdentity);
 390     GNUNET_memcpy (peers,
 391                    req_handle->ids,
 392                    num_peers * sizeof (struct GNUNET_PeerIdentity));
 393     cb_cls = req_handle->cls;
 394     RPS_sampler_request_cancel (req_handle);
 395     req_handle = NULL;
 396     tmp_cb (peers, num_peers, cb_cls);
 397     GNUNET_free (peers);
 398   }
 399 }
 400
 401
 402 /**
 403  * Get n random peers out of the sampled peers.
 404  *
 405  * We might want to reinitialise this sampler after giving the
 406  * corrsponding peer to the client.
 407  * Random with or without consumption?
 408  *
 409  * @param sampler the sampler to get peers from.
 410  * @param cb callback that will be called once the ids are ready.
 411  * @param cls closure given to @a cb
 412  * @param for_client #GNUNET_YES if result is used for client,
 413  *                   #GNUNET_NO if used internally
 414  * @param num_peers the number of peers requested
 415  */
 416 struct RPS_SamplerRequestHandle *
 417 RPS_sampler_get_n_rand_peers (struct RPS_Sampler *sampler,
 418                               uint32_t num_peers,
 419                               RPS_sampler_n_rand_peers_ready_cb cb,
 420                               void *cls)
 421 {
 422   uint32_t i;
 423   struct RPS_SamplerRequestHandle *req_handle;
 424   struct GetPeerCls *gpc;
 425
 426   GNUNET_assert (0 != sampler->sampler_size);
 427   if (0 == num_peers)
 428     return NULL;
 429
 430   // TODO check if we have too much (distinct) sampled peers
 431   req_handle = GNUNET_new (struct RPS_SamplerRequestHandle);
 432   req_handle->num_peers = num_peers;
 433   req_handle->cur_num_peers = 0;
 434   req_handle->ids = GNUNET_new_array (num_peers, struct GNUNET_PeerIdentity);
 435   req_handle->sampler = sampler;
 436   req_handle->callback = cb;
 437   req_handle->cls = cls;
 438   GNUNET_CONTAINER_DLL_insert (sampler->req_handle_head,
 439                                sampler->req_handle_tail,
 440                                req_handle);
 441
 442   LOG (GNUNET_ERROR_TYPE_DEBUG,
 443       "Scheduling requests for %" PRIu32 " peers\n", num_peers);
 444
 445   for (i = 0; i < num_peers; i++)
 446   {
 447     gpc = GNUNET_new (struct GetPeerCls);
 448     gpc->req_handle = req_handle;
 449     gpc->cont = check_n_peers_ready;
 450     gpc->cont_cls = req_handle;
 451     gpc->id = &req_handle->ids[i];
 452
 453     GNUNET_CONTAINER_DLL_insert (req_handle->gpc_head,
 454                                  req_handle->gpc_tail,
 455                                  gpc);
 456     // maybe add a little delay
 457     gpc->get_peer_task = GNUNET_SCHEDULER_add_now (sampler->get_peers,
 458                                                    gpc);
 459   }
 460   return req_handle;
 461 }
 462
 463 /**
 464  * Cancle a request issued through #RPS_sampler_n_rand_peers_ready_cb.
 465  *
 466  * @param req_handle the handle to the request
 467  */
 468 void
 469 RPS_sampler_request_cancel (struct RPS_SamplerRequestHandle *req_handle)
 470 {
 471   struct GetPeerCls *i;
 472
 473   while (NULL != (i = req_handle->gpc_head) )
 474   {
 475     GNUNET_CONTAINER_DLL_remove (req_handle->gpc_head,
 476                                  req_handle->gpc_tail,
 477                                  i);
 478     if (NULL != i->get_peer_task)
 479     {
 480       GNUNET_SCHEDULER_cancel (i->get_peer_task);
 481     }
 482     if (NULL != i->notify_ctx)
 483     {
 484       GNUNET_CONTAINER_DLL_remove (req_handle->sampler->notify_ctx_head,
 485                                    req_handle->sampler->notify_ctx_tail,
 486                                    i->notify_ctx);
 487       GNUNET_free (i->notify_ctx);
 488       i->notify_ctx = NULL;
 489     }
 490     GNUNET_free (i);
 491   }
 492   GNUNET_free (req_handle->ids);
 493   req_handle->ids = NULL;
 494   GNUNET_CONTAINER_DLL_remove (req_handle->sampler->req_handle_head,
 495                                req_handle->sampler->req_handle_tail,
 496                                req_handle);
 497   GNUNET_free (req_handle);
 498 }
 499
 500
 501 /**
 502  * Cleans the sampler.
 503  */
 504   void
 505 RPS_sampler_destroy (struct RPS_Sampler *sampler)
 506 {
 507   if (NULL != sampler->req_handle_head)
 508   {
 509     LOG (GNUNET_ERROR_TYPE_WARNING,
 510         "There are still pending requests. Going to remove them.\n");
 511     while (NULL != sampler->req_handle_head)
 512     {
 513       RPS_sampler_request_cancel (sampler->req_handle_head);
 514     }
 515   }
 516   sampler_empty (sampler);
 517   GNUNET_free (sampler);
 518 }
 519
 520
 521 /* end of rps-sampler_common.c */