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