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