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