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