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 #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 enum RPS_SamplerEmpty
  56 {
  57   NOT_EMPTY = 0x0,
  58       EMPTY = 0x1
  59 };
  60
  61 /**
  62  * A sampler element sampling one PeerID at a time.
  63  */
  64 struct RPS_SamplerElement
  65 {
  66   /**
  67    * Min-wise linear permutation used by this sampler.
  68    *
  69    * This is an key later used by a hmac.
  70    */
  71   struct GNUNET_CRYPTO_AuthKey auth_key;
  72
  73   /**
  74    * The PeerID this sampler currently samples.
  75    */
  76   struct GNUNET_PeerIdentity peer_id;
  77
  78   /**
  79    * The according hash value of this PeerID.
  80    */
  81   struct GNUNET_HashCode peer_id_hash;
  82
  83
  84   /**
  85    * Time of last request.
  86    */
  87   struct GNUNET_TIME_Absolute last_client_request;
  88
  89   /**
  90    * Flag that indicates that we are not holding a valid PeerID right now.
  91    */
  92   enum RPS_SamplerEmpty is_empty;
  93
  94   /**
  95    * 'Birth'
  96    */
  97   struct GNUNET_TIME_Absolute birth;
  98
  99   /**
 100    * How many times a PeerID was put in this sampler.
 101    */
 102   uint32_t num_peers;
 103
 104   /**
 105    * How many times this sampler changed the peer_id.
 106    */
 107   uint32_t num_change;
 108
 109   /**
 110    * The file name this sampler element should log to
 111    */
 112   #ifdef TO_FILE
 113   char *file_name;
 114   #endif /* TO_FILE */
 115 };
 116
 117 /**
 118  * Callback that is called from _get_rand_peer() when the PeerID is ready.
 119  *
 120  * @param cls the closure given alongside this function.
 121  * @param id the PeerID that was returned
 122  */
 123 typedef void
 124 (*RPS_sampler_rand_peer_ready_cont) (void *cls,
 125                                      const struct GNUNET_PeerIdentity *id);
 126
 127
 128 /**
 129  * Closure for #sampler_mod_get_rand_peer() and #sampler_get_rand_peer
 130  */
 131 struct GetPeerCls
 132 {
 133   /**
 134    * DLL
 135    */
 136   struct GetPeerCls *next;
 137
 138   /**
 139    * DLL
 140    */
 141   struct GetPeerCls *prev;
 142
 143   /**
 144    * The sampler this function operates on.
 145    */
 146   struct RPS_Sampler *sampler;
 147
 148   /**
 149    * The task for this function.
 150    */
 151   struct GNUNET_SCHEDULER_Task *get_peer_task;
 152
 153   /**
 154    * The callback
 155    */
 156   RPS_sampler_rand_peer_ready_cont cont;
 157
 158   /**
 159    * The closure to the callback @e cont
 160    */
 161   void *cont_cls;
 162
 163   /**
 164    * The address of the id to be stored at
 165    */
 166   struct GNUNET_PeerIdentity *id;
 167 };
 168
 169
 170 /**
 171  * Type of function used to differentiate between modified and not modified
 172  * Sampler.
 173  */
 174 typedef void
 175 (*RPS_get_peers_type) (void *cls,
 176                        const struct GNUNET_SCHEDULER_TaskContext *tc);
 177
 178 /**
 179  * Get one random peer out of the sampled peers.
 180  *
 181  * We might want to reinitialise this sampler after giving the
 182  * corrsponding peer to the client.
 183  * Only used internally
 184  */
 185 static void
 186 sampler_get_rand_peer (void *cls,
 187                         const struct GNUNET_SCHEDULER_TaskContext *tc);
 188
 189 /**
 190  * Get one random peer out of the sampled peers.
 191  *
 192  * We might want to reinitialise this sampler after giving the
 193  * corrsponding peer to the client.
 194  */
 195 static void
 196 sampler_mod_get_rand_peer (void *cls,
 197                        const struct GNUNET_SCHEDULER_TaskContext *tc);
 198
 199
 200 /**
 201  * Sampler with its own array of SamplerElements
 202  */
 203 struct RPS_Sampler
 204 {
 205   /**
 206    * Number of sampler elements we hold.
 207    */
 208   unsigned int sampler_size;
 209   //size_t size;
 210
 211   /**
 212    * All sampler elements in one array.
 213    */
 214   struct RPS_SamplerElement **sampler_elements;
 215
 216   /**
 217    * Number of sampler elements trash can holds.
 218    */
 219   unsigned int trash_can_size;
 220
 221   /**
 222    * Trash can for old sampler elements.
 223    * We need this to evaluate the sampler.
 224    * TODO remove after evaluation
 225    *      and undo changes in
 226    *      sampler_resize
 227    *      sampler_empty
 228    *      sampler_init
 229    *      sampler_remove?
 230    *      sampler_reinitialise_by_value
 231    *      sampler_update
 232    */
 233   struct RPS_SamplerElement **trash_can;
 234
 235   /**
 236    * Maximum time a round takes
 237    *
 238    * Used in the context of RPS
 239    */
 240   struct GNUNET_TIME_Relative max_round_interval;
 241
 242   /**
 243    * Stores the function to return peers. Which one it is depends on whether
 244    * the Sampler is the modified one or not.
 245    */
 246   RPS_get_peers_type get_peers;
 247
 248   /**
 249    * Head for the DLL to store the closures to pending requests.
 250    */
 251   struct GetPeerCls *gpc_head;
 252
 253   /**
 254    * Tail for the DLL to store the closures to pending requests.
 255    */
 256   struct GetPeerCls *gpc_tail;
 257
 258   #ifdef TO_FILE
 259   /**
 260    * File name to log to
 261    */
 262   char *file_name;
 263   #endif /* TO_FILE */
 264 };
 265
 266 /**
 267  * Closure to _get_n_rand_peers_ready_cb()
 268  */
 269 struct NRandPeersReadyCls
 270 {
 271   /**
 272    * Number of peers we are waiting for.
 273    */
 274   uint32_t num_peers;
 275
 276   /**
 277    * Number of peers we currently have.
 278    */
 279   uint32_t cur_num_peers;
 280
 281   /**
 282    * Pointer to the array holding the ids.
 283    */
 284   struct GNUNET_PeerIdentity *ids;
 285
 286   /**
 287    * Callback to be called when all ids are available.
 288    */
 289   RPS_sampler_n_rand_peers_ready_cb callback;
 290
 291   /**
 292    * Closure given to the callback
 293    */
 294   void *cls;
 295 };
 296
 297 ///**
 298 // * Global sampler variable.
 299 // */
 300 //struct RPS_Sampler *sampler;
 301
 302
 303 /**
 304  * The minimal size for the extended sampler elements.
 305  */
 306 static size_t min_size;
 307
 308 /**
 309  * The maximal size the extended sampler elements should grow to.
 310  */
 311 static size_t max_size;
 312
 313 /**
 314  * The size the extended sampler elements currently have.
 315  */
 316 //static size_t extra_size;
 317
 318 /**
 319  * Inedex to the sampler element that is the next to be returned
 320  */
 321 static uint32_t client_get_index;
 322
 323
 324 /**
 325  * Callback to _get_rand_peer() used by _get_n_rand_peers().
 326  *
 327  * Checks whether all n peers are available. If they are,
 328  * give those back.
 329  */
 330 static void
 331 check_n_peers_ready (void *cls,
 332     const struct GNUNET_PeerIdentity *id)
 333 {
 334   struct NRandPeersReadyCls *n_peers_cls = cls;
 335
 336   n_peers_cls->cur_num_peers++;
 337   LOG (GNUNET_ERROR_TYPE_DEBUG,
 338       "Got %" PRIX32 ". of %" PRIX32 " peers\n",
 339       n_peers_cls->cur_num_peers, n_peers_cls->num_peers);
 340
 341   if (n_peers_cls->num_peers == n_peers_cls->cur_num_peers)
 342   { /* All peers are ready -- return those to the client */
 343     GNUNET_assert (NULL != n_peers_cls->callback);
 344
 345     LOG (GNUNET_ERROR_TYPE_DEBUG,
 346         "returning %" PRIX32 " peers to the client\n",
 347         n_peers_cls->num_peers);
 348     n_peers_cls->callback (n_peers_cls->cls, n_peers_cls->ids, n_peers_cls->num_peers);
 349
 350     GNUNET_free (n_peers_cls);
 351   }
 352 }
 353
 354
 355 /**
 356  * Reinitialise a previously initialised sampler element.
 357  *
 358  * @param sampler pointer to the memory that keeps the value.
 359  */
 360   static void
 361 RPS_sampler_elem_reinit (struct RPS_SamplerElement *sampler_el)
 362 {
 363   sampler_el->is_empty = EMPTY;
 364
 365   // I guess I don't need to call GNUNET_CRYPTO_hmac_derive_key()...
 366   GNUNET_CRYPTO_random_block(GNUNET_CRYPTO_QUALITY_STRONG,
 367                              &(sampler_el->auth_key.key),
 368                              GNUNET_CRYPTO_HASH_LENGTH);
 369
 370   #ifdef TO_FILE
 371   /* Create a file(-name) to store internals to */
 372   char *name_buf;
 373   name_buf = auth_key_to_string (sampler_el->auth_key);
 374
 375   sampler_el->file_name = create_file (name_buf);
 376   GNUNET_free (name_buf);
 377   #endif /* TO_FILE */
 378
 379   sampler_el->last_client_request = GNUNET_TIME_UNIT_FOREVER_ABS;
 380
 381   sampler_el->birth = GNUNET_TIME_absolute_get ();
 382   sampler_el->num_peers = 0;
 383   sampler_el->num_change = 0;
 384 }
 385
 386
 387 /**
 388  * (Re)Initialise given Sampler with random min-wise independent function.
 389  *
 390  * In this implementation this means choosing an auth_key for later use in
 391  * a hmac at random.
 392  *
 393  * @return a newly created RPS_SamplerElement which currently holds no id.
 394  */
 395   struct RPS_SamplerElement *
 396 RPS_sampler_elem_create (void)
 397 {
 398   struct RPS_SamplerElement *s;
 399
 400   s = GNUNET_new (struct RPS_SamplerElement);
 401
 402   RPS_sampler_elem_reinit (s);
 403
 404   return s;
 405 }
 406
 407
 408 /**
 409  * Input an PeerID into the given sampler element.
 410  *
 411  * @param sampler the sampler the @a s_elem belongs to.
 412  *                Needed to know the
 413  */
 414 static void
 415 RPS_sampler_elem_next (struct RPS_SamplerElement *s_elem,
 416                        struct RPS_Sampler *sampler, /* TODO remove? */
 417                        const struct GNUNET_PeerIdentity *other)
 418 {
 419   struct GNUNET_HashCode other_hash;
 420
 421   s_elem->num_peers++;
 422
 423   to_file (s_elem->file_name,
 424            "Got id %s",
 425            GNUNET_i2s_full (other));
 426
 427   if (0 == GNUNET_CRYPTO_cmp_peer_identity (other, &(s_elem->peer_id)))
 428   {
 429     LOG (GNUNET_ERROR_TYPE_DEBUG, "         Got PeerID %s\n",
 430         GNUNET_i2s (other));
 431     LOG (GNUNET_ERROR_TYPE_DEBUG, "Have already PeerID %s\n",
 432         GNUNET_i2s (&(s_elem->peer_id)));
 433   }
 434   else
 435   {
 436     GNUNET_CRYPTO_hmac(&s_elem->auth_key,
 437         other,
 438         sizeof(struct GNUNET_PeerIdentity),
 439         &other_hash);
 440
 441     if (EMPTY == s_elem->is_empty)
 442     {
 443       LOG (GNUNET_ERROR_TYPE_DEBUG,
 444            "Got PeerID %s; Simply accepting (was empty previously).\n",
 445            GNUNET_i2s(other));
 446       s_elem->peer_id = *other;
 447       s_elem->peer_id_hash = other_hash;
 448
 449       s_elem->num_change++;
 450     }
 451     else if (0 > GNUNET_CRYPTO_hash_cmp (&other_hash, &s_elem->peer_id_hash))
 452     {
 453       LOG (GNUNET_ERROR_TYPE_DEBUG, "           Got PeerID %s\n",
 454           GNUNET_i2s (other));
 455       LOG (GNUNET_ERROR_TYPE_DEBUG, "Discarding old PeerID %s\n",
 456           GNUNET_i2s (&s_elem->peer_id));
 457       s_elem->peer_id = *other;
 458       s_elem->peer_id_hash = other_hash;
 459
 460       s_elem->num_change++;
 461     }
 462     else
 463     {
 464       LOG (GNUNET_ERROR_TYPE_DEBUG, "        Got PeerID %s\n",
 465           GNUNET_i2s (other));
 466       LOG (GNUNET_ERROR_TYPE_DEBUG, "Keeping old PeerID %s\n",
 467           GNUNET_i2s (&s_elem->peer_id));
 468     }
 469   }
 470   s_elem->is_empty = NOT_EMPTY;
 471
 472   to_file (s_elem->file_name,
 473            "Now holding %s",
 474            GNUNET_i2s_full (&s_elem->peer_id));
 475 }
 476
 477
 478 /**
 479  * Get the size of the sampler.
 480  *
 481  * @param sampler the sampler to return the size of.
 482  * @return the size of the sampler
 483  */
 484 unsigned int
 485 RPS_sampler_get_size (struct RPS_Sampler *sampler)
 486 {
 487   return sampler->sampler_size;
 488 }
 489
 490
 491 /**
 492  * Grow or shrink the size of the sampler.
 493  *
 494  * @param sampler the sampler to resize.
 495  * @param new_size the new size of the sampler
 496  */
 497 static void
 498 sampler_resize (struct RPS_Sampler *sampler, unsigned int new_size)
 499 {
 500   unsigned int old_size;
 501   uint32_t i;
 502
 503   // TODO check min and max size
 504
 505   old_size = sampler->sampler_size;
 506
 507   if (old_size > new_size)
 508   { /* Shrinking */
 509
 510     LOG (GNUNET_ERROR_TYPE_DEBUG,
 511          "Shrinking sampler %d -> %d\n",
 512          old_size,
 513          new_size);
 514
 515     to_file (sampler->file_name,
 516          "Shrinking sampler %d -> %d",
 517          old_size,
 518          new_size);
 519
 520     /* TODO Temporary store those to properly call the removeCB on those later? */
 521     GNUNET_array_grow (sampler->trash_can,
 522                        sampler->trash_can_size,
 523                        old_size - new_size);
 524     for (i = new_size ; i < old_size ; i++)
 525     {
 526       to_file (sampler->file_name,
 527                "-%" PRIu32 ": %s",
 528                i,
 529                sampler->sampler_elements[i]->file_name);
 530       to_file (sampler->sampler_elements[i]->file_name,
 531                "--- non-active");
 532       sampler->trash_can[i - new_size] = sampler->sampler_elements[i];
 533     }
 534
 535     GNUNET_array_grow (sampler->sampler_elements,
 536                        sampler->sampler_size,
 537                        new_size);
 538     LOG (GNUNET_ERROR_TYPE_DEBUG,
 539          "sampler->sampler_elements now points to %p\n",
 540          sampler->sampler_elements);
 541
 542   }
 543   else if (old_size < new_size)
 544   { /* Growing */
 545     LOG (GNUNET_ERROR_TYPE_DEBUG,
 546          "Growing sampler %d -> %d\n",
 547          old_size,
 548          new_size);
 549
 550     to_file (sampler->file_name,
 551          "Growing sampler %d -> %d",
 552          old_size,
 553          new_size);
 554
 555     GNUNET_array_grow (sampler->sampler_elements,
 556         sampler->sampler_size,
 557         new_size);
 558
 559     for (i = old_size ; i < new_size ; i++)
 560     { /* Add new sampler elements */
 561       sampler->sampler_elements[i] = RPS_sampler_elem_create ();
 562
 563       to_file (sampler->file_name,
 564                "+%" PRIu32 ": %s",
 565                i,
 566                sampler->sampler_elements[i]->file_name);
 567     }
 568   }
 569   else
 570   {
 571     LOG (GNUNET_ERROR_TYPE_DEBUG, "Size remains the same -- nothing to do\n");
 572     return;
 573   }
 574
 575   GNUNET_assert (sampler->sampler_size == new_size);
 576 }
 577
 578
 579 /**
 580  * Grow or shrink the size of the sampler.
 581  *
 582  * @param sampler the sampler to resize.
 583  * @param new_size the new size of the sampler
 584  */
 585 void
 586 RPS_sampler_resize (struct RPS_Sampler *sampler, unsigned int new_size)
 587 {
 588   GNUNET_assert (0 < new_size);
 589   sampler_resize (sampler, new_size);
 590 }
 591
 592
 593 /**
 594  * Empty the sampler.
 595  *
 596  * @param sampler the sampler to empty.
 597  * @param new_size the new size of the sampler
 598  */
 599 static void
 600 sampler_empty (struct RPS_Sampler *sampler)
 601 {
 602   sampler_resize (sampler, 0);
 603   GNUNET_array_grow (sampler->trash_can,
 604                      sampler->trash_can_size,
 605                      0);
 606 }
 607
 608
 609 /**
 610  * Initialise a tuple of sampler elements.
 611  *
 612  * @param init_size the size the sampler is initialised with
 613  * @param max_round_interval maximum time a round takes
 614  * @return a handle to a sampler that consists of sampler elements.
 615  */
 616 struct RPS_Sampler *
 617 RPS_sampler_init (size_t init_size,
 618                   struct GNUNET_TIME_Relative max_round_interval)
 619 {
 620   struct RPS_Sampler *sampler;
 621
 622   /* Initialise context around extended sampler */
 623   min_size = 10; // TODO make input to _samplers_init()
 624   max_size = 1000; // TODO make input to _samplers_init()
 625
 626   sampler = GNUNET_new (struct RPS_Sampler);
 627
 628   #ifdef TO_FILE
 629   sampler->file_name = create_file ("sampler-");
 630
 631   LOG (GNUNET_ERROR_TYPE_DEBUG,
 632        "Initialised sampler %s\n",
 633        sampler->file_name);
 634   #endif /* TO_FILE */
 635
 636   sampler->sampler_size = 0;
 637   sampler->sampler_elements = NULL;
 638   sampler->trash_can_size = 0;
 639   sampler->trash_can = NULL;
 640   sampler->max_round_interval = max_round_interval;
 641   sampler->get_peers = sampler_get_rand_peer;
 642   sampler->gpc_head = NULL;
 643   sampler->gpc_tail = NULL;
 644   //sampler->sampler_elements = GNUNET_new_array(init_size, struct GNUNET_PeerIdentity);
 645   //GNUNET_array_grow (sampler->sampler_elements, sampler->sampler_size, min_size);
 646   RPS_sampler_resize (sampler, init_size);
 647
 648   client_get_index = 0;
 649
 650   //GNUNET_assert (init_size == sampler->sampler_size);
 651   return sampler;
 652 }
 653
 654 /**
 655  * Initialise a modified tuple of sampler elements.
 656  *
 657  * @param init_size the size the sampler is initialised with
 658  * @param max_round_interval maximum time a round takes
 659  * @return a handle to a sampler that consists of sampler elements.
 660  */
 661 struct RPS_Sampler *
 662 RPS_sampler_mod_init (size_t init_size,
 663                       struct GNUNET_TIME_Relative max_round_interval)
 664 {
 665   struct RPS_Sampler *sampler;
 666
 667   sampler = RPS_sampler_init (init_size, max_round_interval);
 668   sampler->get_peers = sampler_mod_get_rand_peer;
 669
 670   LOG (GNUNET_ERROR_TYPE_DEBUG,
 671        "Initialised modified sampler %s\n",
 672        sampler->file_name);
 673   to_file (sampler->file_name,
 674            "This is a modified sampler");
 675
 676   return sampler;
 677 }
 678
 679
 680 /**
 681  * A fuction to update every sampler in the given list
 682  *
 683  * @param sampler the sampler to update.
 684  * @param id the PeerID that is put in the sampler
 685  */
 686   void
 687 RPS_sampler_update (struct RPS_Sampler *sampler,
 688                     const struct GNUNET_PeerIdentity *id)
 689 {
 690   uint32_t i;
 691
 692   to_file (sampler->file_name,
 693            "Got %s",
 694            GNUNET_i2s_full (id));
 695
 696   for (i = 0 ; i < sampler->sampler_size ; i++)
 697   {
 698     RPS_sampler_elem_next (sampler->sampler_elements[i],
 699                            sampler,
 700                            id);
 701   }
 702
 703   for (i = 0 ; i < sampler->trash_can_size ; i++)
 704   {
 705     RPS_sampler_elem_next (sampler->trash_can[i],
 706                            sampler,
 707                            id);
 708   }
 709 }
 710
 711
 712 /**
 713  * Reinitialise all previously initialised sampler elements with the given value.
 714  *
 715  * Used to get rid of a PeerID.
 716  *
 717  * @param sampler the sampler to reinitialise a sampler element in.
 718  * @param id the id of the sampler elements to update.
 719  */
 720   void
 721 RPS_sampler_reinitialise_by_value (struct RPS_Sampler *sampler,
 722                                    const struct GNUNET_PeerIdentity *id)
 723 {
 724   uint32_t i;
 725   struct RPS_SamplerElement *trash_entry;
 726
 727   for ( i = 0 ; i < sampler->sampler_size ; i++ )
 728   {
 729     if ( 0 == GNUNET_CRYPTO_cmp_peer_identity(id, &(sampler->sampler_elements[i]->peer_id)) )
 730     {
 731       LOG (GNUNET_ERROR_TYPE_DEBUG, "Reinitialising sampler\n");
 732       trash_entry = GNUNET_new (struct RPS_SamplerElement);
 733       *trash_entry = *(sampler->sampler_elements[i]);
 734       GNUNET_array_append (sampler->trash_can,
 735                            sampler->trash_can_size,
 736                            trash_entry);
 737       to_file (trash_entry->file_name,
 738                "--- non-active");
 739       RPS_sampler_elem_reinit (sampler->sampler_elements[i]);
 740     }
 741   }
 742 }
 743
 744
 745 /**
 746  * Get one random peer out of the sampled peers.
 747  *
 748  * We might want to reinitialise this sampler after giving the
 749  * corrsponding peer to the client.
 750  * Only used internally
 751  */
 752 static void
 753 sampler_get_rand_peer (void *cls,
 754                         const struct GNUNET_SCHEDULER_TaskContext *tc)
 755 {
 756   struct GetPeerCls *gpc = cls;
 757   uint32_t r_index;
 758
 759   gpc->get_peer_task = NULL;
 760   if (0 != (tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN))
 761     return;
 762
 763   /**;
 764    * Choose the r_index of the peer we want to return
 765    * at random from the interval of the gossip list
 766    */
 767   r_index = GNUNET_CRYPTO_random_u64 (GNUNET_CRYPTO_QUALITY_STRONG,
 768       gpc->sampler->sampler_size);
 769
 770   if (EMPTY == gpc->sampler->sampler_elements[r_index]->is_empty)
 771   {
 772     //LOG (GNUNET_ERROR_TYPE_DEBUG,
 773     //     "Not returning randomly selected, empty PeerID. - Rescheduling.\n");
 774
 775     /* FIXME no active wait - get notified, when new id arrives?
 776      * Might also be a freshly emptied one. Others might still contain ids.
 777      * Counter?
 778      */
 779     gpc->get_peer_task =
 780       GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_relative_multiply (
 781                                         GNUNET_TIME_UNIT_SECONDS, 0.1),
 782                                     &sampler_get_rand_peer,
 783                                     cls);
 784     return;
 785   }
 786
 787   *gpc->id = gpc->sampler->sampler_elements[r_index]->peer_id;
 788
 789   gpc->cont (gpc->cont_cls, gpc->id);
 790
 791   GNUNET_CONTAINER_DLL_remove (gpc->sampler->gpc_head,
 792                                gpc->sampler->gpc_tail,
 793                                gpc);
 794
 795   GNUNET_free (gpc);
 796 }
 797
 798
 799 /**
 800  * Get one random peer out of the sampled peers.
 801  *
 802  * We might want to reinitialise this sampler after giving the
 803  * corrsponding peer to the client.
 804  */
 805 static void
 806 sampler_mod_get_rand_peer (void *cls,
 807                        const struct GNUNET_SCHEDULER_TaskContext *tc)
 808 {
 809   struct GetPeerCls *gpc = cls;
 810   struct GNUNET_PeerIdentity tmp_id;
 811   unsigned int empty_flag;
 812   struct RPS_SamplerElement *s_elem;
 813   struct GNUNET_TIME_Relative last_request_diff;
 814   uint32_t tmp_client_get_index;
 815
 816   gpc->get_peer_task = NULL;
 817   if (0 != (tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN))
 818     return;
 819
 820   LOG (GNUNET_ERROR_TYPE_DEBUG, "Single peer was requested\n");
 821
 822
 823   /* Store the next #client_get_index to check whether we cycled over the whole list */
 824   if (0 < client_get_index)
 825     tmp_client_get_index = client_get_index - 1;
 826   else
 827     tmp_client_get_index = gpc->sampler->sampler_size - 1;
 828
 829   LOG (GNUNET_ERROR_TYPE_DEBUG,
 830       "sched for later if index reaches %" PRIX32 " (sampler size: %" PRIX32 ").\n",
 831       tmp_client_get_index, gpc->sampler->sampler_size);
 832
 833   do
 834   { /* Get first non empty sampler */
 835     if (tmp_client_get_index == client_get_index)
 836     { /* We once cycled over the whole list */
 837       LOG (GNUNET_ERROR_TYPE_DEBUG, "reached tmp_index %" PRIX32 ".\n",
 838            client_get_index);
 839       GNUNET_assert (NULL == gpc->get_peer_task);
 840       gpc->get_peer_task =
 841         GNUNET_SCHEDULER_add_delayed (gpc->sampler->max_round_interval,
 842                                       &sampler_mod_get_rand_peer,
 843                                       cls);
 844       return;
 845     }
 846
 847     tmp_id = gpc->sampler->sampler_elements[client_get_index]->peer_id;
 848     empty_flag = gpc->sampler->sampler_elements[client_get_index]->is_empty;
 849     RPS_sampler_elem_reinit (gpc->sampler->sampler_elements[client_get_index]);
 850     if (EMPTY != empty_flag)
 851       RPS_sampler_elem_next (gpc->sampler->sampler_elements[client_get_index],
 852                              gpc->sampler,
 853                              &tmp_id);
 854
 855     /* Cycle the #client_get_index one step further */
 856     if ( client_get_index == gpc->sampler->sampler_size - 1 )
 857       client_get_index = 0;
 858     else
 859       client_get_index++;
 860
 861     /* LOG (GNUNET_ERROR_TYPE_DEBUG, "incremented index to %" PRIX32 ".\n",
 862          client_get_index); */
 863   } while (EMPTY == gpc->sampler->sampler_elements[client_get_index]->is_empty);
 864
 865   s_elem = gpc->sampler->sampler_elements[client_get_index];
 866   *gpc->id = s_elem->peer_id;
 867
 868   /* Check whether we may use this sampler to give it back to the client */
 869   if (GNUNET_TIME_UNIT_FOREVER_ABS.abs_value_us != s_elem->last_client_request.abs_value_us)
 870   {
 871     last_request_diff =
 872       GNUNET_TIME_absolute_get_difference (s_elem->last_client_request,
 873                                            GNUNET_TIME_absolute_get ());
 874     /* We're not going to give it back now if it was
 875      * already requested by a client this round */
 876     if (last_request_diff.rel_value_us < gpc->sampler->max_round_interval.rel_value_us)
 877     {
 878       LOG (GNUNET_ERROR_TYPE_DEBUG,
 879           "Last client request on this sampler was less than max round interval ago -- scheduling for later\n");
 880       ///* How many time remains untile the next round has started? */
 881       //inv_last_request_diff =
 882       //  GNUNET_TIME_absolute_get_difference (last_request_diff,
 883       //                                       sampler->max_round_interval);
 884       // add a little delay
 885       /* Schedule it one round later */
 886       GNUNET_assert (NULL == gpc->get_peer_task);
 887       gpc->get_peer_task =
 888         GNUNET_SCHEDULER_add_delayed (gpc->sampler->max_round_interval,
 889                                       &sampler_mod_get_rand_peer,
 890                                       cls);
 891       return;
 892     }
 893     // TODO add other reasons to wait here
 894   }
 895
 896   s_elem->last_client_request = GNUNET_TIME_absolute_get ();
 897
 898   GNUNET_CONTAINER_DLL_remove (gpc->sampler->gpc_head,
 899                                gpc->sampler->gpc_tail,
 900                                gpc);
 901   gpc->cont (gpc->cont_cls, gpc->id);
 902   GNUNET_free (gpc);
 903 }
 904
 905
 906 /**
 907  * Get n random peers out of the sampled peers.
 908  *
 909  * We might want to reinitialise this sampler after giving the
 910  * corrsponding peer to the client.
 911  * Random with or without consumption?
 912  *
 913  * @param sampler the sampler to get peers from.
 914  * @param cb callback that will be called once the ids are ready.
 915  * @param cls closure given to @a cb
 916  * @param for_client #GNUNET_YES if result is used for client,
 917  *                   #GNUNET_NO if used internally
 918  * @param num_peers the number of peers requested
 919  */
 920   void
 921 RPS_sampler_get_n_rand_peers (struct RPS_Sampler *sampler,
 922                               RPS_sampler_n_rand_peers_ready_cb cb,
 923                               void *cls, uint32_t num_peers)
 924 {
 925   GNUNET_assert (0 != sampler->sampler_size);
 926
 927   // TODO check if we have too much (distinct) sampled peers
 928   uint32_t i;
 929   struct NRandPeersReadyCls *cb_cls;
 930   struct GetPeerCls *gpc;
 931
 932   cb_cls = GNUNET_new (struct NRandPeersReadyCls);
 933   cb_cls->num_peers = num_peers;
 934   cb_cls->cur_num_peers = 0;
 935   cb_cls->ids = GNUNET_new_array (num_peers, struct GNUNET_PeerIdentity);
 936   cb_cls->callback = cb;
 937   cb_cls->cls = cls;
 938
 939   LOG (GNUNET_ERROR_TYPE_DEBUG,
 940       "Scheduling requests for %" PRIu32 " peers\n", num_peers);
 941
 942   for (i = 0 ; i < num_peers ; i++)
 943   {
 944     gpc = GNUNET_new (struct GetPeerCls);
 945     gpc->sampler = sampler;
 946     gpc->cont = check_n_peers_ready;
 947     gpc->cont_cls = cb_cls;
 948     gpc->id = &cb_cls->ids[i];
 949
 950     // maybe add a little delay
 951     gpc->get_peer_task = GNUNET_SCHEDULER_add_now (sampler->get_peers, gpc);
 952
 953     GNUNET_CONTAINER_DLL_insert (sampler->gpc_head,
 954                                  sampler->gpc_tail,
 955                                  gpc);
 956   }
 957 }
 958
 959
 960 /**
 961  * Counts how many Samplers currently hold a given PeerID.
 962  *
 963  * @param sampler the sampler to count ids in.
 964  * @param id the PeerID to count.
 965  *
 966  * @return the number of occurrences of id.
 967  */
 968   uint32_t
 969 RPS_sampler_count_id (struct RPS_Sampler *sampler,
 970                       const struct GNUNET_PeerIdentity *id)
 971 {
 972   uint32_t count;
 973   uint32_t i;
 974
 975   count = 0;
 976   for ( i = 0 ; i < sampler->sampler_size ; i++ )
 977   {
 978     if ( 0 == GNUNET_CRYPTO_cmp_peer_identity (&sampler->sampler_elements[i]->peer_id, id)
 979         && EMPTY != sampler->sampler_elements[i]->is_empty)
 980       count++;
 981   }
 982   return count;
 983 }
 984
 985
 986 /**
 987  * Cleans the sampler.
 988  */
 989   void
 990 RPS_sampler_destroy (struct RPS_Sampler *sampler)
 991 {
 992   struct GetPeerCls *i;
 993
 994   for (i = sampler->gpc_head; NULL != i; i = sampler->gpc_head)
 995   {
 996     GNUNET_CONTAINER_DLL_remove (sampler->gpc_head,
 997                                  sampler->gpc_tail,
 998                                  i);
 999     GNUNET_SCHEDULER_cancel (i->get_peer_task);
1000     GNUNET_free (i);
1001   }
1002
1003   sampler_empty (sampler);
1004   GNUNET_free (sampler);
1005 }
1006
1007 /* end of gnunet-service-rps.c */