src/rps/gnunet-service-rps.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.c
  23  * @brief rps service implementation
  24  * @author Julius Bünger
  25  */
  26 #include "platform.h"
  27 #include "gnunet_util_lib.h"
  28 #include "gnunet_cadet_service.h"
  29 #include "gnunet_nse_service.h"
  30 #include "rps.h"
  31
  32 #include "gnunet-service-rps_sampler.h"
  33
  34 #include <math.h>
  35 #include <inttypes.h>
  36
  37 #define LOG(kind, ...) GNUNET_log(kind, __VA_ARGS__)
  38
  39 // TODO modify @brief in every file
  40
  41 // TODO check for overflows
  42
  43 // TODO align message structs
  44
  45 // (TODO api -- possibility of getting weak random peer immideately)
  46
  47 // TODO malicious peer
  48
  49 // TODO connect to friends
  50
  51 // TODO store peers somewhere
  52
  53 // TODO ignore list?
  54
  55 // hist_size_init, hist_size_max
  56
  57 /**
  58  * Our configuration.
  59  */
  60 static const struct GNUNET_CONFIGURATION_Handle *cfg;
  61
  62 /**
  63  * Our own identity.
  64  */
  65 static struct GNUNET_PeerIdentity own_identity;
  66
  67
  68   struct GNUNET_PeerIdentity *
  69 get_rand_peer_ignore_list (const struct GNUNET_PeerIdentity *peer_list, unsigned int size,
  70                            const struct GNUNET_PeerIdentity *ignore_list, unsigned int ignore_size);
  71
  72
  73 /***********************************************************************
  74  * Housekeeping with peers
  75 ***********************************************************************/
  76
  77 /**
  78  * Struct used to store the context of a connected client.
  79  */
  80 struct client_ctx
  81 {
  82   /**
  83    * The message queue to communicate with the client.
  84    */
  85   struct GNUNET_MQ_Handle *mq;
  86 };
  87
  88 /**
  89  * Used to keep track in what lists single peerIDs are.
  90  */
  91 enum PeerFlags
  92 {
  93   PULL_REPLY_PENDING   = 0x01,
  94   IN_OTHER_GOSSIP_LIST = 0x02, // unneeded?
  95   IN_OWN_SAMPLER_LIST  = 0x04, // unneeded?
  96   IN_OWN_GOSSIP_LIST   = 0x08, // unneeded?
  97
  98   /**
  99    * We set this bit when we can be sure the other peer is/was live.
 100    */
 101   VALID                = 0x10
 102 };
 103
 104
 105 /**
 106  * Functions of this type can be used to be stored at a peer for later execution.
 107  */
 108 typedef void (* PeerOp) (void *cls, const struct GNUNET_PeerIdentity *peer);
 109
 110 /**
 111  * Outstanding operation on peer consisting of callback and closure
 112  */
 113 struct PeerOutstandingOp
 114 {
 115   /**
 116    * Callback
 117    */
 118   PeerOp op;
 119
 120   /**
 121    * Closure
 122    */
 123   void *op_cls;
 124 };
 125
 126
 127 /**
 128  * Struct used to keep track of other peer's status
 129  *
 130  * This is stored in a multipeermap.
 131  */
 132 struct PeerContext
 133 {
 134   /**
 135    * In own gossip/sampler list, in other's gossip/sampler list
 136    */
 137   uint32_t peer_flags;
 138
 139   /**
 140    * Message queue open to client
 141    */
 142   struct GNUNET_MQ_Handle *mq;
 143
 144   /**
 145    * Channel open to client.
 146    */
 147   struct GNUNET_CADET_Channel *send_channel;
 148
 149   /**
 150    * Channel open from client.
 151    */
 152   struct GNUNET_CADET_Channel *recv_channel; // unneeded?
 153
 154   /**
 155    * Array of outstanding operations on this peer.
 156    */
 157   struct PeerOutstandingOp *outstanding_ops;
 158
 159   /**
 160    * Number of outstanding operations.
 161    */
 162   unsigned int num_outstanding_ops;
 163   //size_t num_outstanding_ops;
 164
 165   /**
 166    * Handle to the callback given to cadet_ntfy_tmt_rdy()
 167    *
 168    * To be canceled on shutdown.
 169    */
 170   struct GNUNET_CADET_TransmitHandle *is_live_task;
 171
 172   /**
 173    * Identity of the peer
 174    */
 175   struct GNUNET_PeerIdentity peer_id;
 176
 177   /**
 178    * This is pobably followed by 'statistical' data (when we first saw
 179    * him, how did we get his ID, how many pushes (in a timeinterval),
 180    * ...)
 181    */
 182 };
 183
 184 /***********************************************************************
 185  * /Housekeeping with peers
 186 ***********************************************************************/
 187
 188
 189
 190
 191
 192 /***********************************************************************
 193  * Globals
 194 ***********************************************************************/
 195
 196 /**
 197  * Sampler used for the Brahms protocol itself.
 198  */
 199 static struct RPS_Sampler *prot_sampler;
 200
 201 /**
 202  * Sampler used for the clients.
 203  */
 204 static struct RPS_Sampler *client_sampler;
 205
 206 /**
 207  * Set of all peers to keep track of them.
 208  */
 209 static struct GNUNET_CONTAINER_MultiPeerMap *peer_map;
 210
 211
 212 /**
 213  * The gossiped list of peers.
 214  */
 215 static struct GNUNET_PeerIdentity *gossip_list;
 216
 217 /**
 218  * Size of the gossiped list
 219  */
 220 //static unsigned int gossip_list_size;
 221 static uint32_t gossip_list_size;
 222
 223
 224 /**
 225  * The size of sampler we need to be able to satisfy the client's need of
 226  * random peers.
 227  */
 228 static unsigned int sampler_size_client_need;
 229
 230 /**
 231  * The size of sampler we need to be able to satisfy the Brahms protocol's
 232  * need of random peers.
 233  *
 234  * This is directly taken as the #gossip_list_size on update of the
 235  * #gossip_list
 236  *
 237  * This is one minimum size the sampler grows to.
 238  */
 239 static unsigned int sampler_size_est_need;
 240
 241
 242 /**
 243  * Percentage of total peer number in the gossip list
 244  * to send random PUSHes to
 245  */
 246 static float alpha;
 247
 248 /**
 249  * Percentage of total peer number in the gossip list
 250  * to send random PULLs to
 251  */
 252 static float beta;
 253
 254 /**
 255  * The percentage gamma of history updates.
 256  * Simply 1 - alpha - beta
 257  */
 258
 259
 260 /**
 261  * Identifier for the main task that runs periodically.
 262  */
 263 static struct GNUNET_SCHEDULER_Task *do_round_task;
 264
 265 /**
 266  * Time inverval the do_round task runs in.
 267  */
 268 static struct GNUNET_TIME_Relative round_interval;
 269
 270
 271
 272 /**
 273  * List to store peers received through pushes temporary.
 274  *
 275  * TODO -> multipeermap
 276  */
 277 static struct GNUNET_PeerIdentity *push_list;
 278
 279 /**
 280  * Size of the push_list;
 281  */
 282 static unsigned int push_list_size;
 283 //size_t push_list_size;
 284
 285 /**
 286  * List to store peers received through pulls temporary.
 287  *
 288  * TODO -> multipeermap
 289  */
 290 static struct GNUNET_PeerIdentity *pull_list;
 291
 292 /**
 293  * Size of the pull_list;
 294  */
 295 static unsigned int pull_list_size;
 296 //size_t pull_list_size;
 297
 298
 299 /**
 300  * Handler to NSE.
 301  */
 302 static struct GNUNET_NSE_Handle *nse;
 303
 304 /**
 305  * Handler to CADET.
 306  */
 307 static struct GNUNET_CADET_Handle *cadet_handle;
 308
 309
 310 /**
 311  * Request counter.
 312  *
 313  * Only needed in the beginning to check how many of the 64 deltas
 314  * we already have
 315  */
 316 static unsigned int req_counter;
 317
 318 /**
 319  * Time of the last request we received.
 320  *
 321  * Used to compute the expected request rate.
 322  */
 323 static struct GNUNET_TIME_Absolute last_request;
 324
 325 /**
 326  * Size of #request_deltas.
 327  */
 328 #define REQUEST_DELTAS_SIZE 64
 329 static unsigned int request_deltas_size = REQUEST_DELTAS_SIZE;
 330
 331 /**
 332  * Last 64 deltas between requests
 333  */
 334 static struct GNUNET_TIME_Relative request_deltas[REQUEST_DELTAS_SIZE];
 335
 336 /**
 337  * The prediction of the rate of requests
 338  */
 339 static struct GNUNET_TIME_Relative  request_rate;
 340
 341
 342 /**
 343  * List with the peers we sent requests to.
 344  */
 345 struct GNUNET_PeerIdentity *pending_pull_reply_list;
 346
 347 /**
 348  * Size of #pending_pull_reply_list.
 349  */
 350 uint32_t pending_pull_reply_list_size;
 351
 352
 353 /**
 354  * Number of history update tasks.
 355  */
 356 uint32_t num_hist_update_tasks;
 357
 358
 359 #if ENABLE_MALICIOUS
 360 /**
 361  * Type of malicious peer
 362  *
 363  * 0 Don't act malicious at all - Default
 364  * 1 Try to maximise representation
 365  * 2 Try to partition the network
 366  */
 367 uint32_t mal_type = 0;
 368
 369 /**
 370  * Other malicious peers
 371  */
 372 static struct GNUNET_PeerIdentity *mal_peers;
 373
 374 /**
 375  * Number of other malicious peers
 376  */
 377 static uint32_t num_mal_peers;
 378
 379 /**
 380  * If type is 2 this is the attacked peer
 381  */
 382 static struct GNUNET_PeerIdentity attacked_peer;
 383 #endif /* ENABLE_MALICIOUS */
 384
 385
 386 /***********************************************************************
 387  * /Globals
 388 ***********************************************************************/
 389
 390
 391
 392
 393
 394
 395 /***********************************************************************
 396  * Util functions
 397 ***********************************************************************/
 398
 399 /**
 400  * Set a peer flag of given peer context.
 401  */
 402 #define set_peer_flag(peer_ctx, mask) (peer_ctx->peer_flags |= mask)
 403
 404 /**
 405  * Get peer flag of given peer context.
 406  */
 407 #define get_peer_flag(peer_ctx, mask) (peer_ctx->peer_flags & mask ? GNUNET_YES : GNUNET_NO)
 408
 409 /**
 410  * Unset flag of given peer context.
 411  */
 412 #define unset_peer_flag(peer_ctx, mask) (peer_ctx->peer_flags &= (~mask))
 413
 414
 415 /**
 416  * Clean the send channel of a peer
 417  */
 418 void
 419 peer_clean (const struct GNUNET_PeerIdentity *peer);
 420
 421
 422 /**
 423  * Check if peer is already in peer array.
 424  */
 425   int
 426 in_arr (const struct GNUNET_PeerIdentity *array,
 427         unsigned int arr_size,
 428         const struct GNUNET_PeerIdentity *peer)
 429 {
 430   GNUNET_assert (NULL != peer);
 431
 432   if (0 == arr_size)
 433     return GNUNET_NO;
 434
 435   GNUNET_assert (NULL != array);
 436
 437   unsigned int i;
 438
 439   for (i = 0; i < arr_size ; i++)
 440     if (0 == GNUNET_CRYPTO_cmp_peer_identity (&array[i], peer))
 441       return GNUNET_YES;
 442   return GNUNET_NO;
 443 }
 444
 445
 446 /**
 447  * Print peerlist to log.
 448  */
 449 void
 450 print_peer_list (struct GNUNET_PeerIdentity *list, unsigned int len)
 451 {
 452   unsigned int i;
 453
 454   LOG (GNUNET_ERROR_TYPE_DEBUG,
 455        "Printing peer list of length %u at %p:\n",
 456        len,
 457        list);
 458   for (i = 0 ; i < len ; i++)
 459   {
 460     LOG (GNUNET_ERROR_TYPE_DEBUG,
 461          "%u. peer: %s\n",
 462          i, GNUNET_i2s (&list[i]));
 463   }
 464 }
 465
 466
 467 /**
 468  * Remove peer from list.
 469  */
 470   void
 471 rem_from_list (struct GNUNET_PeerIdentity **peer_list,
 472                unsigned int *list_size,
 473                const struct GNUNET_PeerIdentity *peer)
 474 {
 475   unsigned int i;
 476   struct GNUNET_PeerIdentity *tmp;
 477
 478   tmp = *peer_list;
 479
 480   LOG (GNUNET_ERROR_TYPE_DEBUG,
 481        "Removing peer %s from list at %p\n",
 482        GNUNET_i2s (peer),
 483        tmp);
 484
 485   for ( i = 0 ; i < *list_size ; i++ )
 486   {
 487     if (0 == GNUNET_CRYPTO_cmp_peer_identity (&tmp[i], peer))
 488     {
 489       if (i < *list_size -1)
 490       { /* Not at the last entry -- shift peers left */
 491         memcpy (&tmp[i], &tmp[i +1],
 492                 ((*list_size) - i -1) * sizeof (struct GNUNET_PeerIdentity));
 493       }
 494       /* Remove last entry (should be now useless PeerID) */
 495       GNUNET_array_grow (tmp, *list_size, (*list_size) -1);
 496     }
 497   }
 498   *peer_list = tmp;
 499 }
 500
 501 /**
 502  * Get random peer from the given list but don't return one from the @a ignore_list.
 503  */
 504   struct GNUNET_PeerIdentity *
 505 get_rand_peer_ignore_list (const struct GNUNET_PeerIdentity *peer_list,
 506                            uint32_t list_size,
 507                            const struct GNUNET_PeerIdentity *ignore_list,
 508                            uint32_t ignore_size)
 509 {
 510   uint32_t r_index;
 511   uint32_t tmp_size;
 512   struct GNUNET_PeerIdentity *tmp_peer_list;
 513   struct GNUNET_PeerIdentity *peer;
 514
 515   GNUNET_assert (NULL != peer_list);
 516   if (0 == list_size)
 517     return NULL;
 518
 519   tmp_size = 0;
 520   tmp_peer_list = NULL;
 521   GNUNET_array_grow (tmp_peer_list, tmp_size, list_size);
 522   memcpy (tmp_peer_list,
 523           peer_list,
 524           list_size * sizeof (struct GNUNET_PeerIdentity));
 525   peer = GNUNET_new (struct GNUNET_PeerIdentity);
 526
 527   /**;
 528    * Choose the r_index of the peer we want to return
 529    * at random from the interval of the gossip list
 530    */
 531   r_index = GNUNET_CRYPTO_random_u64 (GNUNET_CRYPTO_QUALITY_STRONG,
 532                                       tmp_size);
 533   *peer = tmp_peer_list[r_index];
 534
 535   while (in_arr (ignore_list, ignore_size, peer))
 536   {
 537     rem_from_list (&tmp_peer_list, &tmp_size, peer);
 538
 539     print_peer_list (tmp_peer_list, tmp_size);
 540
 541     if (0 == tmp_size)
 542     {
 543       GNUNET_free (peer);
 544       return NULL;
 545     }
 546
 547     /**;
 548      * Choose the r_index of the peer we want to return
 549      * at random from the interval of the gossip list
 550      */
 551     r_index = GNUNET_CRYPTO_random_u64 (GNUNET_CRYPTO_QUALITY_STRONG,
 552                                         tmp_size);
 553     *peer = tmp_peer_list[r_index];
 554   }
 555
 556
 557   GNUNET_array_grow (tmp_peer_list, tmp_size, 0);
 558
 559   return peer;
 560 }
 561
 562
 563 /**
 564  * Get the context of a peer. If not existing, create.
 565  */
 566   struct PeerContext *
 567 get_peer_ctx (struct GNUNET_CONTAINER_MultiPeerMap *peer_map,
 568               const struct GNUNET_PeerIdentity *peer)
 569 {
 570   struct PeerContext *ctx;
 571
 572   if ( GNUNET_YES == GNUNET_CONTAINER_multipeermap_contains (peer_map, peer))
 573   {
 574     ctx = GNUNET_CONTAINER_multipeermap_get (peer_map, peer);
 575   }
 576   else
 577   {
 578     ctx = GNUNET_new (struct PeerContext);
 579     ctx->peer_flags = 0;
 580     ctx->mq = NULL;
 581     ctx->send_channel = NULL;
 582     ctx->recv_channel = NULL;
 583     ctx->outstanding_ops = NULL;
 584     ctx->num_outstanding_ops = 0;
 585     ctx->is_live_task = NULL;
 586     ctx->peer_id = *peer;
 587     (void) GNUNET_CONTAINER_multipeermap_put (peer_map, peer, ctx,
 588                                               GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_FAST);
 589   }
 590   return ctx;
 591 }
 592
 593
 594 /**
 595  * Put random peer from sampler into the gossip list as history update.
 596  */
 597   void
 598 hist_update (void *cls, struct GNUNET_PeerIdentity *ids, uint32_t num_peers)
 599 {
 600   GNUNET_assert (1 == num_peers);
 601
 602   if (gossip_list_size < sampler_size_est_need)
 603     GNUNET_array_append (gossip_list, gossip_list_size, *ids);
 604
 605   if (0 < num_hist_update_tasks)
 606     num_hist_update_tasks--;
 607 }
 608
 609
 610 /**
 611  * Set the peer flag to living and call the outstanding operations on this peer.
 612  */
 613 static size_t
 614 peer_is_live (struct PeerContext *peer_ctx)
 615 {
 616   struct GNUNET_PeerIdentity *peer;
 617
 618   peer = &peer_ctx->peer_id;
 619   set_peer_flag (peer_ctx, VALID);
 620
 621   LOG (GNUNET_ERROR_TYPE_DEBUG, "Peer %s is live\n", GNUNET_i2s (peer));
 622
 623   if (0 != peer_ctx->num_outstanding_ops)
 624   { /* Call outstanding operations */
 625     unsigned int i;
 626
 627     for (i = 0 ; i < peer_ctx->num_outstanding_ops ; i++)
 628       peer_ctx->outstanding_ops[i].op (peer_ctx->outstanding_ops[i].op_cls, peer);
 629     GNUNET_array_grow (peer_ctx->outstanding_ops, peer_ctx->num_outstanding_ops, 0);
 630   }
 631
 632   return 0;
 633 }
 634
 635
 636 /**
 637  * Callback that is called when a channel was effectively established.
 638  * This is given to ntfy_tmt_rdy and called when the channel was
 639  * successfully established.
 640  */
 641 static size_t
 642 cadet_ntfy_tmt_rdy_cb (void *cls, size_t size, void *buf)
 643 {
 644   struct PeerContext *peer_ctx = (struct PeerContext *) cls;
 645
 646   if (NULL != buf
 647       && 0 != size)
 648     peer_is_live (peer_ctx);
 649
 650   //if (NULL != peer_ctx->is_live_task)
 651   //{
 652   //  LOG (GNUNET_ERROR_TYPE_DEBUG,
 653   //       "Trying to cancle is_live_task for peer %s\n",
 654   //       GNUNET_i2s (&peer_ctx->peer_id));
 655   //  GNUNET_CADET_notify_transmit_ready_cancel (peer_ctx->is_live_task);
 656   //  peer_ctx->is_live_task = NULL;
 657   //}
 658   peer_ctx->is_live_task = NULL;
 659
 660   return 0;
 661 }
 662
 663
 664 /**
 665  * Get the channel of a peer. If not existing, create.
 666  */
 667   struct GNUNET_CADET_Channel *
 668 get_channel (struct GNUNET_CONTAINER_MultiPeerMap *peer_map,
 669              const struct GNUNET_PeerIdentity *peer)
 670 {
 671   struct PeerContext *peer_ctx;
 672
 673   LOG (GNUNET_ERROR_TYPE_DEBUG,
 674        "Trying to establish channel to peer %s\n",
 675        GNUNET_i2s (peer));
 676
 677   peer_ctx = get_peer_ctx (peer_map, peer);
 678   if (NULL == peer_ctx->send_channel)
 679   {
 680     peer_ctx->send_channel =
 681       GNUNET_CADET_channel_create (cadet_handle,
 682                                    NULL,
 683                                    peer,
 684                                    GNUNET_RPS_CADET_PORT,
 685                                    GNUNET_CADET_OPTION_RELIABLE);
 686
 687     // do I have to explicitly put it in the peer_map?
 688     (void) GNUNET_CONTAINER_multipeermap_put
 689       (peer_map,
 690        peer,
 691        peer_ctx,
 692        GNUNET_CONTAINER_MULTIHASHMAPOPTION_REPLACE);
 693   }
 694   return peer_ctx->send_channel;
 695 }
 696
 697
 698 /**
 699  * Get the message queue of a specific peer.
 700  *
 701  * If we already have a message queue open to this client,
 702  * simply return it, otherways create one.
 703  */
 704   struct GNUNET_MQ_Handle *
 705 get_mq (struct GNUNET_CONTAINER_MultiPeerMap *peer_map,
 706         const struct GNUNET_PeerIdentity *peer_id)
 707 {
 708   struct PeerContext *peer_ctx;
 709
 710   peer_ctx = get_peer_ctx (peer_map, peer_id);
 711
 712   if (NULL == peer_ctx->mq)
 713   {
 714     (void) get_channel (peer_map, peer_id);
 715     peer_ctx->mq = GNUNET_CADET_mq_create (peer_ctx->send_channel);
 716     //do I have to explicitly put it in the peer_map?
 717     (void) GNUNET_CONTAINER_multipeermap_put (peer_map, peer_id, peer_ctx,
 718                                        GNUNET_CONTAINER_MULTIHASHMAPOPTION_REPLACE);
 719   }
 720   return peer_ctx->mq;
 721 }
 722
 723
 724 /**
 725  * Issue check whether peer is live
 726  *
 727  * @param peer_ctx the context of the peer
 728  */
 729 void
 730 check_peer_live (struct PeerContext *peer_ctx)
 731 {
 732   (void) get_channel (peer_map, &peer_ctx->peer_id);
 733   LOG (GNUNET_ERROR_TYPE_DEBUG,
 734        "Get informed about peer %s getting live\n",
 735        GNUNET_i2s (&peer_ctx->peer_id));
 736   peer_ctx->is_live_task =
 737       GNUNET_CADET_notify_transmit_ready (peer_ctx->send_channel,
 738                                           GNUNET_NO,
 739                                           GNUNET_TIME_UNIT_FOREVER_REL,
 740                                           sizeof (struct GNUNET_MessageHeader),
 741                                           cadet_ntfy_tmt_rdy_cb,
 742                                           peer_ctx);
 743   // FIXME check whether this is NULL
 744 }
 745
 746
 747 /**
 748  * Sum all time relatives of an array.
 749   */
 750   struct GNUNET_TIME_Relative
 751 T_relative_sum (const struct GNUNET_TIME_Relative *rel_array, uint32_t arr_size)
 752 {
 753   struct GNUNET_TIME_Relative sum;
 754   uint32_t i;
 755
 756   sum = GNUNET_TIME_UNIT_ZERO;
 757   for ( i = 0 ; i < arr_size ; i++ )
 758   {
 759     sum = GNUNET_TIME_relative_add (sum, rel_array[i]);
 760   }
 761   return sum;
 762 }
 763
 764
 765 /**
 766  * Compute the average of given time relatives.
 767  */
 768   struct GNUNET_TIME_Relative
 769 T_relative_avg (const struct GNUNET_TIME_Relative *rel_array, uint32_t arr_size)
 770 {
 771   return GNUNET_TIME_relative_divide (T_relative_sum (rel_array, arr_size), arr_size);
 772 }
 773
 774
 775 /**
 776  * Insert PeerID in #pull_list
 777  *
 778  * Called once we know a peer is live.
 779  */
 780   void
 781 insert_in_pull_list (void *cls, const struct GNUNET_PeerIdentity *peer)
 782 {
 783   if (GNUNET_NO == in_arr (pull_list, pull_list_size, peer))
 784     GNUNET_array_append (pull_list, pull_list_size, *peer);
 785
 786   peer_clean (peer);
 787 }
 788
 789 /**
 790  * Check whether #insert_in_pull_list was already scheduled
 791  */
 792   int
 793 insert_in_pull_list_scheduled (const struct PeerContext *peer_ctx)
 794 {
 795   unsigned int i;
 796
 797   for ( i = 0 ; i < peer_ctx->num_outstanding_ops ; i++ )
 798     if (insert_in_pull_list == peer_ctx->outstanding_ops[i].op)
 799       return GNUNET_YES;
 800   return GNUNET_NO;
 801 }
 802
 803
 804 /**
 805  * Insert PeerID in #gossip_list
 806  *
 807  * Called once we know a peer is live.
 808  */
 809   void
 810 insert_in_gossip_list (void *cls, const struct GNUNET_PeerIdentity *peer)
 811 {
 812   if (GNUNET_NO == in_arr (gossip_list, gossip_list_size, peer))
 813     GNUNET_array_append (gossip_list, gossip_list_size, *peer);
 814
 815   (void) get_channel (peer_map, peer);
 816 }
 817
 818 /**
 819  * Check whether #insert_in_gossip_list was already scheduled
 820  */
 821   int
 822 insert_in_gossip_list_scheduled (const struct PeerContext *peer_ctx)
 823 {
 824   unsigned int i;
 825
 826   for ( i = 0 ; i < peer_ctx->num_outstanding_ops ; i++ )
 827     if (insert_in_gossip_list == peer_ctx->outstanding_ops[i].op)
 828       return GNUNET_YES;
 829   return GNUNET_NO;
 830 }
 831
 832
 833 /**
 834  * Update sampler with given PeerID.
 835  */
 836   void
 837 insert_in_sampler (void *cls, const struct GNUNET_PeerIdentity *peer)
 838 {
 839   LOG (GNUNET_ERROR_TYPE_DEBUG,
 840        "Updating samplers with peer %s from insert_in_sampler()\n",
 841        GNUNET_i2s (peer));
 842   RPS_sampler_update (prot_sampler,   peer);
 843   RPS_sampler_update (client_sampler, peer);
 844 }
 845
 846
 847 /**
 848  * Check whether #insert_in_sampler was already scheduled
 849  */
 850 static int
 851 insert_in_sampler_scheduled (const struct PeerContext *peer_ctx)
 852 {
 853   unsigned int i;
 854
 855   for ( i = 0 ; i < peer_ctx->num_outstanding_ops ; i++ )
 856     if (insert_in_sampler== peer_ctx->outstanding_ops[i].op)
 857       return GNUNET_YES;
 858   return GNUNET_NO;
 859 }
 860
 861
 862 /**
 863  * Wrapper around #RPS_sampler_resize()
 864  *
 865  * If we do not have enough sampler elements, double current sampler size
 866  * If we have more than enough sampler elements, halv current sampler size
 867  */
 868 static void
 869 resize_wrapper (struct RPS_Sampler *sampler, uint32_t new_size)
 870 {
 871   unsigned int sampler_size;
 872
 873   // TODO statistics
 874   // TODO respect the min, max
 875   sampler_size = RPS_sampler_get_size (sampler);
 876   if (sampler_size > new_size * 4)
 877   { /* Shrinking */
 878     RPS_sampler_resize (sampler, sampler_size / 2);
 879   }
 880   else if (sampler_size < new_size)
 881   { /* Growing */
 882     RPS_sampler_resize (sampler, sampler_size * 2);
 883   }
 884   LOG (GNUNET_ERROR_TYPE_DEBUG, "sampler_size is now %u\n", sampler_size);
 885 }
 886
 887
 888 /**
 889  * Wrapper around #RPS_sampler_resize() resizing the client sampler
 890  */
 891 static void
 892 client_resize_wrapper ()
 893 {
 894   uint32_t bigger_size;
 895   unsigned int sampler_size;
 896
 897   // TODO statistics
 898
 899   sampler_size = RPS_sampler_get_size (client_sampler);
 900
 901   if (sampler_size_est_need > sampler_size_client_need)
 902     bigger_size = sampler_size_est_need;
 903   else
 904     bigger_size = sampler_size_client_need;
 905
 906   // TODO respect the min, max
 907   resize_wrapper (client_sampler, bigger_size);
 908   LOG (GNUNET_ERROR_TYPE_DEBUG, "sampler_size is now %u\n", sampler_size);
 909 }
 910
 911
 912 /**
 913  * Estimate request rate
 914  *
 915  * Called every time we receive a request from the client.
 916  */
 917   void
 918 est_request_rate()
 919 {
 920   struct GNUNET_TIME_Relative max_round_duration;
 921
 922   if (request_deltas_size > req_counter)
 923     req_counter++;
 924   if ( 1 < req_counter)
 925   {
 926     /* Shift last request deltas to the right */
 927     memcpy (&request_deltas[1],
 928         request_deltas,
 929         (req_counter - 1) * sizeof (struct GNUNET_TIME_Relative));
 930
 931     /* Add current delta to beginning */
 932     request_deltas[0] =
 933         GNUNET_TIME_absolute_get_difference (last_request,
 934                                              GNUNET_TIME_absolute_get ());
 935     request_rate = T_relative_avg (request_deltas, req_counter);
 936
 937     /* Compute the duration a round will maximally take */
 938     max_round_duration =
 939         GNUNET_TIME_relative_add (round_interval,
 940                                   GNUNET_TIME_relative_divide (round_interval, 2));
 941
 942     /* Set the estimated size the sampler has to have to
 943      * satisfy the current client request rate */
 944     sampler_size_client_need =
 945         max_round_duration.rel_value_us / request_rate.rel_value_us;
 946
 947     /* Resize the sampler */
 948     client_resize_wrapper ();
 949   }
 950   last_request = GNUNET_TIME_absolute_get ();
 951 }
 952
 953
 954 /***********************************************************************
 955  * /Util functions
 956 ***********************************************************************/
 957
 958
 959
 960
 961
 962 /**
 963  * Function called by NSE.
 964  *
 965  * Updates sizes of sampler list and gossip list and adapt those lists
 966  * accordingly.
 967  */
 968   void
 969 nse_callback (void *cls, struct GNUNET_TIME_Absolute timestamp,
 970               double logestimate, double std_dev)
 971 {
 972   double estimate;
 973   //double scale; // TODO this might go gloabal/config
 974
 975   LOG (GNUNET_ERROR_TYPE_DEBUG,
 976        "Received a ns estimate - logest: %f, std_dev: %f (old_size: %u)\n",
 977        logestimate, std_dev, RPS_sampler_get_size (prot_sampler));
 978   //scale = .01;
 979   estimate = GNUNET_NSE_log_estimate_to_n (logestimate);
 980   // GNUNET_NSE_log_estimate_to_n (logestimate);
 981   estimate = pow (estimate, 1.0 / 3);
 982   // TODO add if std_dev is a number
 983   // estimate += (std_dev * scale);
 984   if (2 < ceil (estimate))
 985   {
 986     LOG (GNUNET_ERROR_TYPE_DEBUG, "Changing estimate to %f\n", estimate);
 987     sampler_size_est_need = estimate;
 988   } else
 989     LOG (GNUNET_ERROR_TYPE_DEBUG, "Not using estimate %f\n", estimate);
 990
 991   /* If the NSE has changed adapt the lists accordingly */
 992   resize_wrapper (prot_sampler, sampler_size_est_need);
 993   client_resize_wrapper ();
 994 }
 995
 996
 997 /**
 998  * Callback called once the requested PeerIDs are ready.
 999  *
1000  * Sends those to the requesting client.
1001  */
1002 void client_respond (void *cls,
1003     struct GNUNET_PeerIdentity *ids, uint32_t num_peers)
1004 {
1005   LOG (GNUNET_ERROR_TYPE_DEBUG, "sampler returned %" PRIX32 " peers\n", num_peers);
1006   struct GNUNET_MQ_Envelope *ev;
1007   struct GNUNET_RPS_CS_ReplyMessage *out_msg;
1008   struct GNUNET_SERVER_Client *client;
1009   uint32_t size_needed;
1010   struct client_ctx *cli_ctx;
1011
1012   client = (struct GNUNET_SERVER_Client *) cls;
1013
1014   size_needed = sizeof (struct GNUNET_RPS_CS_ReplyMessage) +
1015                 num_peers * sizeof (struct GNUNET_PeerIdentity);
1016
1017   GNUNET_assert (GNUNET_SERVER_MAX_MESSAGE_SIZE >= size_needed);
1018
1019   ev = GNUNET_MQ_msg_extra (out_msg,
1020                             num_peers * sizeof (struct GNUNET_PeerIdentity),
1021                             GNUNET_MESSAGE_TYPE_RPS_CS_REPLY);
1022   out_msg->num_peers = htonl (num_peers);
1023
1024   memcpy (&out_msg[1],
1025       ids,
1026       num_peers * sizeof (struct GNUNET_PeerIdentity));
1027   GNUNET_free (ids);
1028
1029   cli_ctx = GNUNET_SERVER_client_get_user_context (client, struct client_ctx);
1030   if ( NULL == cli_ctx ) {
1031     cli_ctx = GNUNET_new (struct client_ctx);
1032     cli_ctx->mq = GNUNET_MQ_queue_for_server_client (client);
1033     GNUNET_SERVER_client_set_user_context (client, cli_ctx);
1034   }
1035
1036   GNUNET_MQ_send (cli_ctx->mq, ev);
1037 }
1038
1039
1040 /**
1041  * Handle RPS request from the client.
1042  *
1043  * @param cls closure
1044  * @param client identification of the client
1045  * @param message the actual message
1046  */
1047 static void
1048 handle_client_request (void *cls,
1049             struct GNUNET_SERVER_Client *client,
1050             const struct GNUNET_MessageHeader *message)
1051 {
1052   struct GNUNET_RPS_CS_RequestMessage *msg;
1053   uint32_t num_peers;
1054   uint32_t size_needed;
1055   uint32_t i;
1056
1057   msg = (struct GNUNET_RPS_CS_RequestMessage *) message;
1058
1059   num_peers = ntohl (msg->num_peers);
1060   size_needed = sizeof (struct GNUNET_RPS_CS_RequestMessage) +
1061                 num_peers * sizeof (struct GNUNET_PeerIdentity);
1062
1063   if (GNUNET_SERVER_MAX_MESSAGE_SIZE < size_needed)
1064   {
1065     GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
1066     return;
1067   }
1068
1069   for (i = 0 ; i < num_peers ; i++)
1070     est_request_rate();
1071
1072   LOG (GNUNET_ERROR_TYPE_DEBUG, "Client requested %" PRIX32 " random peer(s).\n", num_peers);
1073
1074   RPS_sampler_get_n_rand_peers (client_sampler, client_respond,
1075                                 client, num_peers, GNUNET_YES);
1076
1077   GNUNET_SERVER_receive_done (client,
1078                               GNUNET_OK);
1079 }
1080
1081
1082 /**
1083  * Handle seed from the client.
1084  *
1085  * @param cls closure
1086  * @param client identification of the client
1087  * @param message the actual message
1088  */
1089   static void
1090 handle_client_seed (void *cls,
1091             struct GNUNET_SERVER_Client *client,
1092             const struct GNUNET_MessageHeader *message)
1093 {
1094   struct GNUNET_RPS_CS_SeedMessage *in_msg;
1095   struct GNUNET_PeerIdentity *peers;
1096   uint32_t num_peers;
1097   uint32_t i;
1098
1099   if (sizeof (struct GNUNET_RPS_CS_SeedMessage) > ntohs (message->size))
1100   {
1101     GNUNET_break_op (0);
1102     GNUNET_SERVER_receive_done (client,
1103                                 GNUNET_SYSERR);
1104   }
1105
1106   in_msg = (struct GNUNET_RPS_CS_SeedMessage *) message;
1107   num_peers = ntohl (in_msg->num_peers);
1108   peers = (struct GNUNET_PeerIdentity *) &in_msg[1];
1109   //peers = GNUNET_new_array (num_peers, struct GNUNET_PeerIdentity);
1110   //memcpy (peers, &in_msg[1], num_peers * sizeof (struct GNUNET_PeerIdentity));
1111
1112   if ((ntohs (message->size) - sizeof (struct GNUNET_RPS_CS_SeedMessage)) /
1113       sizeof (struct GNUNET_PeerIdentity) != num_peers)
1114   {
1115     GNUNET_break_op (0);
1116     GNUNET_SERVER_receive_done (client,
1117                                 GNUNET_SYSERR);
1118   }
1119
1120   LOG (GNUNET_ERROR_TYPE_DEBUG,
1121        "Client seeded peers:\n");
1122   print_peer_list (peers, num_peers);
1123
1124   // TODO check for validity of ids
1125
1126   for (i = 0 ; i < num_peers ; i++)
1127   {
1128     LOG (GNUNET_ERROR_TYPE_DEBUG,
1129          "Updating samplers with seed %" PRIX32 ": %s\n",
1130          i,
1131          GNUNET_i2s (&peers[i]));
1132
1133     RPS_sampler_update (prot_sampler,   &peers[i]);
1134     RPS_sampler_update (client_sampler, &peers[i]);
1135   }
1136
1137   //GNUNET_free (peers);
1138
1139   GNUNET_SERVER_receive_done (client,
1140                                                 GNUNET_OK);
1141 }
1142
1143
1144 /**
1145  * Handle a PUSH message from another peer.
1146  *
1147  * Check the proof of work and store the PeerID
1148  * in the temporary list for pushed PeerIDs.
1149  *
1150  * @param cls Closure
1151  * @param channel The channel the PUSH was received over
1152  * @param channel_ctx The context associated with this channel
1153  * @param msg The message header
1154  */
1155 static int
1156 handle_peer_push (void *cls,
1157     struct GNUNET_CADET_Channel *channel,
1158     void **channel_ctx,
1159     const struct GNUNET_MessageHeader *msg)
1160 {
1161   const struct GNUNET_PeerIdentity *peer;
1162
1163   // (check the proof of work)
1164
1165   peer = (const struct GNUNET_PeerIdentity *) GNUNET_CADET_channel_get_info (channel, GNUNET_CADET_OPTION_PEER);
1166   // FIXME wait for cadet to change this function
1167   LOG (GNUNET_ERROR_TYPE_DEBUG, "PUSH received (%s)\n", GNUNET_i2s (peer));
1168
1169   /* Add the sending peer to the push_list */
1170   if (GNUNET_NO == in_arr (push_list, push_list_size, peer))
1171     GNUNET_array_append (push_list, push_list_size, *peer);
1172
1173   return GNUNET_OK;
1174 }
1175
1176 /**
1177  * Handle PULL REQUEST request message from another peer.
1178  *
1179  * Reply with the gossip list of PeerIDs.
1180  *
1181  * @param cls Closure
1182  * @param channel The channel the PUSH was received over
1183  * @param channel_ctx The context associated with this channel
1184  * @param msg The message header
1185  */
1186 static int
1187 handle_peer_pull_request (void *cls,
1188     struct GNUNET_CADET_Channel *channel,
1189     void **channel_ctx,
1190     const struct GNUNET_MessageHeader *msg)
1191 {
1192   struct GNUNET_PeerIdentity *peer;
1193   uint32_t send_size;
1194   struct GNUNET_MQ_Handle *mq;
1195   struct GNUNET_MQ_Envelope *ev;
1196   struct GNUNET_RPS_P2P_PullReplyMessage *out_msg;
1197
1198
1199   peer = (struct GNUNET_PeerIdentity *) GNUNET_CADET_channel_get_info (channel,
1200                                                                        GNUNET_CADET_OPTION_PEER);
1201   // FIXME wait for cadet to change this function
1202
1203   /* Compute actual size */
1204   send_size = sizeof (struct GNUNET_RPS_P2P_PullReplyMessage) +
1205               gossip_list_size * sizeof (struct GNUNET_PeerIdentity);
1206
1207   if (GNUNET_CONSTANTS_MAX_CADET_MESSAGE_SIZE < send_size)
1208     /* Compute number of peers to send
1209      * If too long, simply truncate */
1210     send_size =
1211       (GNUNET_CONSTANTS_MAX_CADET_MESSAGE_SIZE -
1212        sizeof (struct GNUNET_RPS_P2P_PullReplyMessage)) /
1213        sizeof (struct GNUNET_PeerIdentity);
1214   else
1215     send_size = gossip_list_size;
1216
1217   LOG (GNUNET_ERROR_TYPE_DEBUG,
1218       "PULL REQUEST from peer %s received, going to send %u peers\n",
1219       GNUNET_i2s (peer), send_size);
1220
1221   mq = get_mq (peer_map, peer);
1222
1223   ev = GNUNET_MQ_msg_extra (out_msg,
1224                            send_size * sizeof (struct GNUNET_PeerIdentity),
1225                            GNUNET_MESSAGE_TYPE_RPS_PP_PULL_REPLY);
1226   //out_msg->num_peers = htonl (gossip_list_size);
1227   out_msg->num_peers = htonl (send_size);
1228   memcpy (&out_msg[1], gossip_list,
1229          send_size * sizeof (struct GNUNET_PeerIdentity));
1230
1231   GNUNET_MQ_send (mq, ev);
1232
1233   return GNUNET_OK;
1234 }
1235
1236
1237 /**
1238  * Handle PULL REPLY message from another peer.
1239  *
1240  * Check whether we sent a corresponding request and
1241  * whether this reply is the first one.
1242  *
1243  * @param cls Closure
1244  * @param channel The channel the PUSH was received over
1245  * @param channel_ctx The context associated with this channel
1246  * @param msg The message header
1247  */
1248   static int
1249 handle_peer_pull_reply (void *cls,
1250                         struct GNUNET_CADET_Channel *channel,
1251                         void **channel_ctx,
1252                         const struct GNUNET_MessageHeader *msg)
1253 {
1254   LOG (GNUNET_ERROR_TYPE_DEBUG, "PULL REPLY received\n");
1255
1256   struct GNUNET_RPS_P2P_PullReplyMessage *in_msg;
1257   struct GNUNET_PeerIdentity *peers;
1258   struct PeerContext *peer_ctx;
1259   struct GNUNET_PeerIdentity *sender;
1260   struct PeerContext *sender_ctx;
1261   struct PeerOutstandingOp out_op;
1262   uint32_t i;
1263
1264   /* Check for protocol violation */
1265   if (sizeof (struct GNUNET_RPS_P2P_PullReplyMessage) > ntohs (msg->size))
1266   {
1267     GNUNET_break_op (0);
1268     return GNUNET_SYSERR;
1269   }
1270
1271   in_msg = (struct GNUNET_RPS_P2P_PullReplyMessage *) msg;
1272   if ((ntohs (msg->size) - sizeof (struct GNUNET_RPS_P2P_PullReplyMessage)) /
1273       sizeof (struct GNUNET_PeerIdentity) != ntohl (in_msg->num_peers))
1274   {
1275     LOG (GNUNET_ERROR_TYPE_ERROR,
1276         "message says it sends %" PRIu64 " peers, have space for %i peers\n",
1277         ntohl (in_msg->num_peers),
1278         (ntohs (msg->size) - sizeof (struct GNUNET_RPS_P2P_PullReplyMessage)) /
1279             sizeof (struct GNUNET_PeerIdentity));
1280     GNUNET_break_op (0);
1281     return GNUNET_SYSERR;
1282   }
1283
1284   sender = (struct GNUNET_PeerIdentity *) GNUNET_CADET_channel_get_info (
1285       (struct GNUNET_CADET_Channel *) channel, GNUNET_CADET_OPTION_PEER);
1286        // Guess simply casting isn't the nicest way...
1287        // FIXME wait for cadet to change this function
1288   sender_ctx = get_peer_ctx (peer_map, sender);
1289
1290   if (GNUNET_YES == get_peer_flag (sender_ctx, PULL_REPLY_PENDING))
1291   {
1292     GNUNET_break_op (0);
1293     return GNUNET_OK;
1294   }
1295
1296   /* Do actual logic */
1297   peers = (struct GNUNET_PeerIdentity *) &msg[1];
1298   for (i = 0 ; i < ntohl (in_msg->num_peers) ; i++)
1299   {
1300     peer_ctx = get_peer_ctx (peer_map, &peers[i]);
1301     if (GNUNET_YES == get_peer_flag (peer_ctx, VALID)
1302         || NULL != peer_ctx->send_channel
1303         || NULL != peer_ctx->recv_channel)
1304     {
1305       if (GNUNET_NO == in_arr (pull_list, pull_list_size, &peers[i])
1306           && 0 != GNUNET_CRYPTO_cmp_peer_identity (&own_identity, &peers[i]))
1307         GNUNET_array_append (pull_list, pull_list_size, peers[i]);
1308     }
1309     else if (GNUNET_NO == insert_in_pull_list_scheduled (peer_ctx))
1310     {
1311       out_op.op = insert_in_pull_list;
1312       out_op.op_cls = NULL;
1313       GNUNET_array_append (peer_ctx->outstanding_ops,
1314                            peer_ctx->num_outstanding_ops,
1315                            out_op);
1316       check_peer_live (peer_ctx);
1317     }
1318   }
1319
1320   unset_peer_flag (sender_ctx, PULL_REPLY_PENDING);
1321   rem_from_list (&pending_pull_reply_list, &pending_pull_reply_list_size, sender);
1322
1323   return GNUNET_OK;
1324 }
1325
1326
1327 #if ENABLE_MALICIOUS
1328 /**
1329  * Turn RPS service to act malicious.
1330  *
1331  * @param cls Closure
1332  * @param channel The channel the PUSH was received over
1333  * @param channel_ctx The context associated with this channel
1334  * @param msg The message header
1335  */
1336   static int
1337 handle_peer_act_malicious (void *cls,
1338                            struct GNUNET_CADET_Channel *channel,
1339                            void **channel_ctx,
1340                            const struct GNUNET_MessageHeader *msg)
1341 {
1342   struct GNUNET_RPS_CS_ActMaliciousMessage *in_msg;
1343   struct GNUNET_PeerIdentity *sender;
1344   struct PeerContext *sender_ctx;
1345   struct GNUNET_PeerIdentity *peers;
1346
1347   /* Check for protocol violation */
1348   if (sizeof (struct GNUNET_RPS_CS_ActMaliciousMessage) > ntohs (msg->size))
1349   {
1350     GNUNET_break_op (0);
1351     return GNUNET_SYSERR;
1352   }
1353
1354   in_msg = (struct GNUNET_RPS_CS_ActMaliciousMessage *) msg;
1355   if ((ntohs (msg->size) - sizeof (struct GNUNET_RPS_CS_ActMaliciousMessage)) /
1356       sizeof (struct GNUNET_PeerIdentity) != ntohl (in_msg->num_peers))
1357   {
1358     LOG (GNUNET_ERROR_TYPE_ERROR,
1359         "message says it sends %" PRIu64 " peers, have space for %i peers\n",
1360         ntohl (in_msg->num_peers),
1361         (ntohs (msg->size) - sizeof (struct GNUNET_RPS_CS_ActMaliciousMessage)) /
1362             sizeof (struct GNUNET_PeerIdentity));
1363     GNUNET_break_op (0);
1364     return GNUNET_SYSERR;
1365   }
1366
1367   sender = (struct GNUNET_PeerIdentity *) GNUNET_CADET_channel_get_info (
1368       (struct GNUNET_CADET_Channel *) channel, GNUNET_CADET_OPTION_PEER);
1369        // Guess simply casting isn't the nicest way...
1370        // FIXME wait for cadet to change this function
1371   sender_ctx = get_peer_ctx (peer_map, sender);
1372
1373   if (GNUNET_YES == get_peer_flag (sender_ctx, PULL_REPLY_PENDING))
1374   {
1375     GNUNET_break_op (0);
1376     return GNUNET_OK;
1377   }
1378
1379
1380   /* Do actual logic */
1381   peers = (struct GNUNET_PeerIdentity *) &msg[1];
1382   num_mal_peers = ntohl (in_msg->num_peers);
1383   mal_type = ntohl (in_msg->type);
1384
1385   LOG (GNUNET_ERROR_TYPE_DEBUG,
1386        "Now acting malicious type %" PRIX32 "\n",
1387        mal_type);
1388
1389   if (1 == mal_type)
1390   { /* Try to maximise representation */
1391     num_mal_peers = ntohl (in_msg->num_peers);
1392     mal_peers = GNUNET_new_array (num_mal_peers,
1393                                   struct GNUNET_PeerIdentity);
1394     memcpy (mal_peers, peers, num_mal_peers);
1395   }
1396   else if (2 == mal_type)
1397   { /* Try to partition the network */
1398     num_mal_peers = ntohl (in_msg->num_peers) - 1;
1399     mal_peers = GNUNET_new_array (num_mal_peers,
1400                                   struct GNUNET_PeerIdentity);
1401     memcpy (mal_peers, peers, num_mal_peers);
1402     attacked_peer = peers[num_mal_peers];
1403   }
1404
1405   return GNUNET_OK;
1406 }
1407
1408
1409 /**
1410  * Send out PUSHes and PULLs maliciously.
1411  *
1412  * This is executed regylary.
1413  */
1414 static void
1415 do_mal_round (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
1416 {
1417   LOG (GNUNET_ERROR_TYPE_DEBUG, "Going to execute next round maliciously.\n");
1418
1419   /* Do stuff */
1420
1421   /* Compute random time value between .5 * round_interval and 1.5 *round_interval */
1422   half_round_interval = GNUNET_TIME_relative_divide (round_interval, 2);
1423   do
1424   {
1425   /*
1426    * Compute random value between (0 and 1) * round_interval
1427    * via multiplying round_interval with a 'fraction' (0 to value)/value
1428    */
1429   rand_delay = GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK, UINT_MAX/10);
1430   time_next_round = GNUNET_TIME_relative_multiply (round_interval,  rand_delay);
1431   time_next_round = GNUNET_TIME_relative_divide   (time_next_round, UINT_MAX/10);
1432   time_next_round = GNUNET_TIME_relative_add      (time_next_round, half_round_interval);
1433   } while (GNUNET_TIME_UNIT_FOREVER_REL.rel_value_us == time_next_round.rel_value_us);
1434
1435   /* Schedule next round */
1436   do_round_task = GNUNET_SCHEDULER_add_delayed (round_interval, &do_mal_round, NULL);
1437   LOG (GNUNET_ERROR_TYPE_DEBUG, "Finished round\n");
1438 }
1439 #endif /* ENABLE_MALICIOUS */
1440
1441
1442 /**
1443  * Send out PUSHes and PULLs, possibly update #gossip_list, samplers.
1444  *
1445  * This is executed regylary.
1446  */
1447 static void
1448 do_round (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
1449 {
1450   LOG (GNUNET_ERROR_TYPE_DEBUG, "Going to execute next round.\n");
1451
1452   uint32_t i;
1453   unsigned int *permut;
1454   unsigned int n_peers; /* Number of peers we send pushes/pulls to */
1455   struct GNUNET_MQ_Envelope *ev;
1456   struct GNUNET_PeerIdentity peer;
1457   struct GNUNET_PeerIdentity *tmp_peer;
1458   struct GNUNET_MQ_Handle *mq;
1459
1460   LOG (GNUNET_ERROR_TYPE_DEBUG,
1461        "Printing gossip list:\n");
1462   for (i = 0 ; i < gossip_list_size ; i++)
1463     LOG (GNUNET_ERROR_TYPE_DEBUG,
1464          "\t%s\n", GNUNET_i2s (&gossip_list[i]));
1465   // TODO log lists, ...
1466
1467   /* Would it make sense to have one shuffeled gossip list and then
1468    * to send PUSHes to first alpha peers, PULL requests to next beta peers and
1469    * use the rest to update sampler?
1470    * in essence get random peers with consumption */
1471
1472   /* Send PUSHes */
1473   if (0 < gossip_list_size)
1474   {
1475     permut = GNUNET_CRYPTO_random_permute (GNUNET_CRYPTO_QUALITY_STRONG,
1476                                            (unsigned int) gossip_list_size);
1477     n_peers = ceil (alpha * gossip_list_size);
1478     LOG (GNUNET_ERROR_TYPE_DEBUG,
1479          "Going to send pushes to %u ceil (%f * %u) peers.\n",
1480          n_peers, alpha, gossip_list_size);
1481     for (i = 0 ; i < n_peers ; i++)
1482     {
1483       peer = gossip_list[permut[i]];
1484       if (0 != GNUNET_CRYPTO_cmp_peer_identity (&own_identity, &peer)) // TODO
1485       { // FIXME if this fails schedule/loop this for later
1486         LOG (GNUNET_ERROR_TYPE_DEBUG,
1487              "Sending PUSH to peer %s of gossiped list.\n",
1488              GNUNET_i2s (&peer));
1489
1490         ev = GNUNET_MQ_msg_header (GNUNET_MESSAGE_TYPE_RPS_PP_PUSH);
1491         mq = get_mq (peer_map, &peer);
1492         GNUNET_MQ_send (mq, ev);
1493       }
1494     }
1495     GNUNET_free (permut);
1496   }
1497
1498
1499   /* Send PULL requests */
1500   //permut = GNUNET_CRYPTO_random_permute (GNUNET_CRYPTO_QUALITY_STRONG, (unsigned int) sampler_list->size);
1501   n_peers = ceil (beta * gossip_list_size);
1502   LOG (GNUNET_ERROR_TYPE_DEBUG,
1503        "Going to send pulls to %u ceil (%f * %u) peers.\n",
1504        n_peers, beta, gossip_list_size);
1505   for (i = 0 ; i < n_peers ; i++)
1506   {
1507     tmp_peer = get_rand_peer_ignore_list (gossip_list, gossip_list_size,
1508         pending_pull_reply_list, pending_pull_reply_list_size);
1509     if (NULL != tmp_peer)
1510     {
1511       peer = *tmp_peer;
1512       GNUNET_free (tmp_peer);
1513
1514       GNUNET_array_append (pending_pull_reply_list, pending_pull_reply_list_size, peer);
1515
1516       if (0 != GNUNET_CRYPTO_cmp_peer_identity (&own_identity, &peer))
1517       { // FIXME if this fails schedule/loop this for later
1518         LOG (GNUNET_ERROR_TYPE_DEBUG,
1519              "Sending PULL request to peer %s of gossiped list.\n",
1520              GNUNET_i2s (&peer));
1521
1522         ev = GNUNET_MQ_msg_header (GNUNET_MESSAGE_TYPE_RPS_PP_PULL_REQUEST);
1523         mq = get_mq (peer_map, &peer);
1524         GNUNET_MQ_send (mq, ev);
1525       }
1526     }
1527   }
1528
1529
1530   /* Update gossip list */
1531
1532   if ( push_list_size <= alpha * gossip_list_size &&
1533        push_list_size != 0 &&
1534        pull_list_size != 0 )
1535   {
1536     LOG (GNUNET_ERROR_TYPE_DEBUG, "Update of the gossip list.\n");
1537
1538     uint32_t first_border;
1539     uint32_t second_border;
1540     uint32_t r_index;
1541     uint32_t peers_to_clean_size;
1542     struct GNUNET_PeerIdentity *peers_to_clean;
1543
1544     peers_to_clean = NULL;
1545     peers_to_clean_size = 0;
1546     GNUNET_array_grow (peers_to_clean, peers_to_clean_size, gossip_list_size);
1547     memcpy (peers_to_clean,
1548             gossip_list,
1549             gossip_list_size * sizeof (struct GNUNET_PeerIdentity));
1550
1551     first_border  =                ceil (alpha * sampler_size_est_need);
1552     second_border = first_border + ceil (beta  * sampler_size_est_need);
1553
1554     GNUNET_array_grow (gossip_list, gossip_list_size, second_border);
1555
1556     for (i = 0 ; i < first_border ; i++)
1557     { // TODO use RPS_sampler_get_n_rand_peers
1558       /* Update gossip list with peers received through PUSHes */
1559       r_index = GNUNET_CRYPTO_random_u64 (GNUNET_CRYPTO_QUALITY_STRONG,
1560                                        push_list_size);
1561       gossip_list[i] = push_list[r_index];
1562       // TODO change the peer_flags accordingly
1563     }
1564
1565     for (i = first_border ; i < second_border ; i++)
1566     {
1567       /* Update gossip list with peers received through PULLs */
1568       r_index = GNUNET_CRYPTO_random_u64 (GNUNET_CRYPTO_QUALITY_STRONG,
1569                                        pull_list_size);
1570       gossip_list[i] = pull_list[r_index];
1571       // TODO change the peer_flags accordingly
1572     }
1573
1574     for (i = second_border ; i < sampler_size_est_need ; i++)
1575     {
1576       /* Update gossip list with peers from history */
1577       RPS_sampler_get_n_rand_peers (prot_sampler, hist_update, NULL, 1, GNUNET_NO);
1578       num_hist_update_tasks++;
1579       // TODO change the peer_flags accordingly
1580     }
1581
1582     for (i = 0 ; i < gossip_list_size ; i++)
1583       rem_from_list (&peers_to_clean, &peers_to_clean_size, &gossip_list[i]);
1584
1585     for (i = 0 ; i < peers_to_clean_size ; i++)
1586       peer_clean (&peers_to_clean[i]);
1587
1588     GNUNET_free (peers_to_clean);
1589   }
1590   else
1591   {
1592     LOG (GNUNET_ERROR_TYPE_DEBUG, "No update of the gossip list.\n");
1593   }
1594   // TODO independent of that also get some peers from CADET_get_peers()?
1595
1596
1597   /* Update samplers */
1598   for ( i = 0 ; i < push_list_size ; i++ )
1599   {
1600     LOG (GNUNET_ERROR_TYPE_DEBUG,
1601          "Updating with peer %s from push list\n",
1602          GNUNET_i2s (&push_list[i]));
1603     RPS_sampler_update (prot_sampler,   &push_list[i]);
1604     RPS_sampler_update (client_sampler, &push_list[i]);
1605     // TODO set in_flag?
1606   }
1607
1608   for ( i = 0 ; i < pull_list_size ; i++ )
1609   {
1610     LOG (GNUNET_ERROR_TYPE_DEBUG,
1611          "Updating with peer %s from pull list\n",
1612          GNUNET_i2s (&pull_list[i]));
1613     RPS_sampler_update (prot_sampler,   &push_list[i]);
1614     RPS_sampler_update (client_sampler, &push_list[i]);
1615     // TODO set in_flag?
1616   }
1617
1618
1619   /* Empty push/pull lists */
1620   GNUNET_array_grow (push_list, push_list_size, 0);
1621   GNUNET_array_grow (pull_list, pull_list_size, 0);
1622
1623   struct GNUNET_TIME_Relative time_next_round;
1624   struct GNUNET_TIME_Relative half_round_interval;
1625   unsigned int rand_delay;
1626
1627
1628   /* Compute random time value between .5 * round_interval and 1.5 *round_interval */
1629   half_round_interval = GNUNET_TIME_relative_divide (round_interval, 2);
1630   do
1631   {
1632   /*
1633    * Compute random value between (0 and 1) * round_interval
1634    * via multiplying round_interval with a 'fraction' (0 to value)/value
1635    */
1636   rand_delay = GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK, UINT_MAX/10);
1637   time_next_round = GNUNET_TIME_relative_multiply (round_interval,  rand_delay);
1638   time_next_round = GNUNET_TIME_relative_divide   (time_next_round, UINT_MAX/10);
1639   time_next_round = GNUNET_TIME_relative_add      (time_next_round, half_round_interval);
1640   } while (GNUNET_TIME_UNIT_FOREVER_REL.rel_value_us == time_next_round.rel_value_us);
1641
1642   /* Schedule next round */
1643   do_round_task = GNUNET_SCHEDULER_add_delayed (round_interval, &do_round, NULL);
1644   LOG (GNUNET_ERROR_TYPE_DEBUG, "Finished round\n");
1645 }
1646
1647
1648 static void
1649 rps_start (struct GNUNET_SERVER_Handle *server);
1650
1651
1652 /**
1653  * This is called from GNUNET_CADET_get_peers().
1654  *
1655  * It is called on every peer(ID) that cadet somehow has contact with.
1656  * We use those to initialise the sampler.
1657  */
1658 void
1659 init_peer_cb (void *cls,
1660               const struct GNUNET_PeerIdentity *peer,
1661               int tunnel, // "Do we have a tunnel towards this peer?"
1662               unsigned int n_paths, // "Number of known paths towards this peer"
1663               unsigned int best_path) // "How long is the best path?
1664                                       // (0 = unknown, 1 = ourselves, 2 = neighbor)"
1665 {
1666   struct PeerOutstandingOp out_op;
1667   struct PeerContext *peer_ctx;
1668
1669   if (NULL != peer
1670       && 0 != GNUNET_CRYPTO_cmp_peer_identity (&own_identity, peer))
1671   {
1672     LOG (GNUNET_ERROR_TYPE_DEBUG,
1673         "Got peer %s (at %p) from CADET (gossip_list_size: %u)\n",
1674         GNUNET_i2s (peer), peer, gossip_list_size);
1675
1676     // maybe create a function for that
1677     peer_ctx = get_peer_ctx (peer_map, peer);
1678     if (GNUNET_YES != get_peer_flag (peer_ctx, VALID))
1679     {
1680       if (GNUNET_NO == insert_in_sampler_scheduled (peer_ctx))
1681       {
1682         out_op.op = insert_in_sampler;
1683         out_op.op_cls = NULL;
1684         GNUNET_array_append (peer_ctx->outstanding_ops,
1685                              peer_ctx->num_outstanding_ops,
1686                              out_op);
1687       }
1688
1689       if (GNUNET_NO == insert_in_gossip_list_scheduled (peer_ctx))
1690       {
1691         out_op.op = insert_in_gossip_list;
1692         out_op.op_cls = NULL;
1693         GNUNET_array_append (peer_ctx->outstanding_ops,
1694                              peer_ctx->num_outstanding_ops,
1695                              out_op);
1696       }
1697
1698       /* Trigger livelyness test on peer */
1699       check_peer_live (peer_ctx);
1700     }
1701
1702     // send push/pull to each of those peers?
1703   }
1704 }
1705
1706
1707 /**
1708  * Clean the send channel of a peer
1709  */
1710 void
1711 peer_clean (const struct GNUNET_PeerIdentity *peer)
1712 {
1713   struct PeerContext *peer_ctx;
1714   struct GNUNET_CADET_Channel *channel;
1715
1716   if (GNUNET_YES != in_arr (gossip_list, gossip_list_size, peer)
1717       && GNUNET_YES == GNUNET_CONTAINER_multipeermap_contains (peer_map, peer))
1718   {
1719     peer_ctx = get_peer_ctx (peer_map, peer);
1720     if (NULL != peer_ctx->send_channel)
1721     {
1722       channel = peer_ctx->send_channel;
1723       peer_ctx->send_channel = NULL;
1724       GNUNET_CADET_channel_destroy (channel);
1725     }
1726   }
1727 }
1728
1729
1730 /**
1731  * Callback used to remove peers from the multipeermap.
1732  */
1733   int
1734 peer_remove_cb (void *cls, const struct GNUNET_PeerIdentity *key, void *value)
1735 {
1736   struct PeerContext *peer_ctx;
1737   const struct GNUNET_CADET_Channel *channel =
1738     (const struct GNUNET_CADET_Channel *) cls;
1739   struct GNUNET_CADET_Channel *recv;
1740   struct GNUNET_CADET_Channel *send;
1741
1742   if (GNUNET_YES == GNUNET_CONTAINER_multipeermap_contains (peer_map, value))
1743   {
1744     peer_ctx = (struct PeerContext *) value;
1745
1746     if (0 != peer_ctx->num_outstanding_ops)
1747       GNUNET_array_grow (peer_ctx->outstanding_ops,
1748                          peer_ctx->num_outstanding_ops,
1749                          0);
1750
1751     if (NULL != peer_ctx->mq)
1752       GNUNET_MQ_destroy (peer_ctx->mq);
1753
1754     if (NULL != peer_ctx->is_live_task)
1755     {
1756     LOG (GNUNET_ERROR_TYPE_DEBUG,
1757          "Trying to cancle is_live_task for peer %s\n",
1758          GNUNET_i2s (key));
1759       GNUNET_CADET_notify_transmit_ready_cancel (peer_ctx->is_live_task);
1760       peer_ctx->is_live_task = NULL;
1761     }
1762
1763     send = peer_ctx->send_channel;
1764     peer_ctx->send_channel = NULL;
1765     if (NULL != send
1766         && channel != send)
1767     {
1768       GNUNET_CADET_channel_destroy (send);
1769     }
1770
1771     recv = peer_ctx->send_channel;
1772     peer_ctx->recv_channel = NULL;
1773     if (NULL != recv
1774         && channel != recv)
1775     {
1776       GNUNET_CADET_channel_destroy (recv);
1777     }
1778
1779     if (GNUNET_YES != GNUNET_CONTAINER_multipeermap_remove_all (peer_map, key))
1780       LOG (GNUNET_ERROR_TYPE_WARNING, "removing peer from peer_map failed\n");
1781     else
1782       GNUNET_free (peer_ctx);
1783   }
1784
1785   return GNUNET_YES;
1786 }
1787
1788
1789 /**
1790  * Task run during shutdown.
1791  *
1792  * @param cls unused
1793  * @param tc unused
1794  */
1795 static void
1796 shutdown_task (void *cls,
1797                      const struct GNUNET_SCHEDULER_TaskContext *tc)
1798 {
1799   LOG (GNUNET_ERROR_TYPE_DEBUG, "RPS is going down\n");
1800
1801   if (NULL != do_round_task)
1802   {
1803     GNUNET_SCHEDULER_cancel (do_round_task);
1804     do_round_task = NULL;
1805   }
1806
1807
1808   {
1809   if (GNUNET_SYSERR ==
1810         GNUNET_CONTAINER_multipeermap_iterate (peer_map, peer_remove_cb, NULL))
1811     LOG (GNUNET_ERROR_TYPE_WARNING,
1812         "Iterating over peers to disconnect from them was cancelled\n");
1813   }
1814
1815   GNUNET_NSE_disconnect (nse);
1816   GNUNET_CADET_disconnect (cadet_handle);
1817   RPS_sampler_destroy (prot_sampler);
1818   RPS_sampler_destroy (client_sampler);
1819   LOG (GNUNET_ERROR_TYPE_DEBUG,
1820        "Size of the peermap: %u\n",
1821        GNUNET_CONTAINER_multipeermap_size (peer_map));
1822   GNUNET_break (0 == GNUNET_CONTAINER_multipeermap_size (peer_map));
1823   GNUNET_CONTAINER_multipeermap_destroy (peer_map);
1824   GNUNET_array_grow (gossip_list, gossip_list_size, 0);
1825   GNUNET_array_grow (push_list, push_list_size, 0);
1826   GNUNET_array_grow (pull_list, pull_list_size, 0);
1827 }
1828
1829
1830 /**
1831  * A client disconnected.  Remove all of its data structure entries.
1832  *
1833  * @param cls closure, NULL
1834  * @param client identification of the client
1835  */
1836 static void
1837 handle_client_disconnect (void *cls,
1838                           struct GNUNET_SERVER_Client * client)
1839 {
1840 }
1841
1842
1843 /**
1844  * Handle the channel a peer opens to us.
1845  *
1846  * @param cls The closure
1847  * @param channel The channel the peer wants to establish
1848  * @param initiator The peer's peer ID
1849  * @param port The port the channel is being established over
1850  * @param options Further options
1851  */
1852   static void *
1853 handle_inbound_channel (void *cls,
1854                         struct GNUNET_CADET_Channel *channel,
1855                         const struct GNUNET_PeerIdentity *initiator,
1856                         uint32_t port,
1857                         enum GNUNET_CADET_ChannelOption options)
1858 {
1859   struct PeerContext *peer_ctx;
1860   struct GNUNET_PeerIdentity peer;
1861
1862   peer = *initiator;
1863   LOG (GNUNET_ERROR_TYPE_DEBUG,
1864       "New channel was established to us (Peer %s).\n",
1865       GNUNET_i2s (&peer));
1866
1867   GNUNET_assert (NULL != channel);
1868
1869   // we might not even store the recv_channel
1870
1871   peer_ctx = get_peer_ctx (peer_map, &peer);
1872   // FIXME what do we do if a channel is established twice?
1873   //       overwrite? Clean old channel? ...?
1874   //if (NULL != peer_ctx->recv_channel)
1875   //{
1876   //  peer_ctx->recv_channel = channel;
1877   //}
1878   peer_ctx->recv_channel = channel;
1879
1880   peer_ctx->mq = NULL;
1881
1882   (void) GNUNET_CONTAINER_multipeermap_put (peer_map, &peer, peer_ctx,
1883       GNUNET_CONTAINER_MULTIHASHMAPOPTION_REPLACE);
1884
1885   peer_is_live (peer_ctx);
1886
1887   return NULL; // TODO
1888 }
1889
1890
1891 /**
1892  * This is called when a remote peer destroys a channel.
1893  *
1894  * @param cls The closure
1895  * @param channel The channel being closed
1896  * @param channel_ctx The context associated with this channel
1897  */
1898   static void
1899 cleanup_channel (void *cls,
1900                 const struct GNUNET_CADET_Channel *channel,
1901                 void *channel_ctx)
1902 {
1903   struct GNUNET_PeerIdentity *peer;
1904   struct PeerContext *peer_ctx;
1905
1906   peer = (struct GNUNET_PeerIdentity *) GNUNET_CADET_channel_get_info (
1907       (struct GNUNET_CADET_Channel *) channel, GNUNET_CADET_OPTION_PEER);
1908        // Guess simply casting isn't the nicest way...
1909        // FIXME wait for cadet to change this function
1910   LOG (GNUNET_ERROR_TYPE_DEBUG, "Cleaning up channel to peer %s\n",
1911        GNUNET_i2s (peer));
1912
1913   if (GNUNET_YES == GNUNET_CONTAINER_multipeermap_contains (peer_map, peer))
1914   {
1915     peer_ctx = GNUNET_CONTAINER_multipeermap_get (peer_map, peer);
1916
1917     if (NULL == peer_ctx) /* It could have been removed by shutdown_task */
1918       return;
1919
1920     if (channel == peer_ctx->send_channel)
1921     { /* Peer probably went down */
1922       rem_from_list (&gossip_list, &gossip_list_size, peer);
1923       rem_from_list (&pending_pull_reply_list, &pending_pull_reply_list_size, peer);
1924
1925       /* Somwewhat {ab,re}use the iterator function */
1926       /* Cast to void is ok, because it's used as void in peer_remove_cb */
1927       (void) peer_remove_cb ((void *) channel, peer, peer_ctx);
1928     }
1929     else /* Other peer doesn't want to send us messages anymore */
1930       peer_ctx->recv_channel = NULL;
1931   }
1932 }
1933
1934
1935 /**
1936  * Actually start the service.
1937  */
1938   static void
1939 rps_start (struct GNUNET_SERVER_Handle *server)
1940 {
1941   static const struct GNUNET_SERVER_MessageHandler handlers[] = {
1942     {&handle_client_request, NULL, GNUNET_MESSAGE_TYPE_RPS_CS_REQUEST,
1943       sizeof (struct GNUNET_RPS_CS_RequestMessage)},
1944     {&handle_client_seed,    NULL, GNUNET_MESSAGE_TYPE_RPS_CS_SEED, 0},
1945     {NULL, NULL, 0, 0}
1946   };
1947
1948   GNUNET_SERVER_add_handlers (server, handlers);
1949   GNUNET_SERVER_disconnect_notify (server,
1950                                    &handle_client_disconnect,
1951                                    NULL);
1952   LOG (GNUNET_ERROR_TYPE_DEBUG, "Ready to receive requests from clients\n");
1953
1954
1955   do_round_task = GNUNET_SCHEDULER_add_now (&do_round, NULL);
1956   LOG (GNUNET_ERROR_TYPE_DEBUG, "Scheduled first round\n");
1957
1958   GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_FOREVER_REL,
1959                                                         &shutdown_task,
1960                                                         NULL);
1961 }
1962
1963
1964 /**
1965  * Process statistics requests.
1966  *
1967  * @param cls closure
1968  * @param server the initialized server
1969  * @param c configuration to use
1970  */
1971   static void
1972 run (void *cls,
1973      struct GNUNET_SERVER_Handle *server,
1974      const struct GNUNET_CONFIGURATION_Handle *c)
1975 {
1976   // TODO check what this does -- copied from gnunet-boss
1977   // - seems to work as expected
1978   GNUNET_log_setup ("rps", GNUNET_error_type_to_string (GNUNET_ERROR_TYPE_DEBUG), NULL);
1979   cfg = c;
1980
1981
1982   /* Get own ID */
1983   GNUNET_CRYPTO_get_peer_identity (cfg, &own_identity); // TODO check return value
1984   GNUNET_log (GNUNET_ERROR_TYPE_INFO,
1985               "STARTING SERVICE (rps) for peer [%s]\n",
1986               GNUNET_i2s (&own_identity));
1987
1988
1989   /* Get time interval from the configuration */
1990   if (GNUNET_OK != GNUNET_CONFIGURATION_get_value_time (cfg, "RPS",
1991                                                         "ROUNDINTERVAL",
1992                                                         &round_interval))
1993   {
1994     LOG (GNUNET_ERROR_TYPE_DEBUG, "Failed to read ROUNDINTERVAL from config\n");
1995     GNUNET_SCHEDULER_shutdown ();
1996     return;
1997   }
1998
1999   /* Get initial size of sampler/gossip list from the configuration */
2000   if (GNUNET_OK != GNUNET_CONFIGURATION_get_value_number (cfg, "RPS",
2001                                                          "INITSIZE",
2002                                                          (long long unsigned int *) &sampler_size_est_need))
2003   {
2004     LOG (GNUNET_ERROR_TYPE_DEBUG, "Failed to read INITSIZE from config\n");
2005     GNUNET_SCHEDULER_shutdown ();
2006     return;
2007   }
2008   LOG (GNUNET_ERROR_TYPE_DEBUG, "INITSIZE is %" PRIu64 "\n", sampler_size_est_need);
2009
2010
2011   gossip_list = NULL;
2012
2013
2014   /* connect to NSE */
2015   nse = GNUNET_NSE_connect (cfg, nse_callback, NULL);
2016   // TODO check whether that was successful
2017   LOG (GNUNET_ERROR_TYPE_DEBUG, "Connected to NSE\n");
2018
2019
2020   alpha = 0.45;
2021   beta  = 0.45;
2022
2023   peer_map = GNUNET_CONTAINER_multipeermap_create (sampler_size_est_need, GNUNET_NO);
2024
2025
2026   /* Initialise cadet */
2027   static const struct GNUNET_CADET_MessageHandler cadet_handlers[] = {
2028     {&handle_peer_push        , GNUNET_MESSAGE_TYPE_RPS_PP_PUSH        ,
2029       sizeof (struct GNUNET_MessageHeader)},
2030     {&handle_peer_pull_request, GNUNET_MESSAGE_TYPE_RPS_PP_PULL_REQUEST,
2031       sizeof (struct GNUNET_MessageHeader)},
2032     {&handle_peer_pull_reply  , GNUNET_MESSAGE_TYPE_RPS_PP_PULL_REPLY  , 0},
2033     #if ENABLE_MALICIOUS
2034     {&handle_peer_act_malicious, GNUNET_MESSAGE_TYPE_RPS_ACT_MALICIOUS , 0},
2035     #endif /* ENABLE_MALICIOUS */
2036     {NULL, 0, 0}
2037   };
2038
2039   const uint32_t ports[] = {GNUNET_RPS_CADET_PORT, 0}; // _PORT specified in src/rps/rps.h
2040   cadet_handle = GNUNET_CADET_connect (cfg,
2041                                        cls,
2042                                        &handle_inbound_channel,
2043                                        &cleanup_channel,
2044                                        cadet_handlers,
2045                                        ports);
2046   LOG (GNUNET_ERROR_TYPE_DEBUG, "Connected to CADET\n");
2047
2048
2049   /* Initialise sampler */
2050   struct GNUNET_TIME_Relative half_round_interval;
2051   struct GNUNET_TIME_Relative  max_round_interval;
2052
2053   half_round_interval = GNUNET_TIME_relative_multiply (round_interval, .5);
2054   max_round_interval = GNUNET_TIME_relative_add (round_interval, half_round_interval);
2055
2056   prot_sampler =   RPS_sampler_init (sampler_size_est_need, max_round_interval);
2057   client_sampler = RPS_sampler_init (sampler_size_est_need, max_round_interval);
2058
2059   /* Initialise push and pull maps */
2060   push_list = NULL;
2061   push_list_size = 0;
2062   pull_list = NULL;
2063   pull_list_size = 0;
2064   pending_pull_reply_list = NULL;
2065   pending_pull_reply_list_size = 0;
2066
2067
2068   num_hist_update_tasks = 0;
2069
2070
2071   LOG (GNUNET_ERROR_TYPE_DEBUG, "Requesting peers from CADET\n");
2072   GNUNET_CADET_get_peers (cadet_handle, &init_peer_cb, NULL);
2073   // TODO send push/pull to each of those peers?
2074
2075
2076   rps_start (server);
2077 }
2078
2079
2080 /**
2081  * The main function for the rps service.
2082  *
2083  * @param argc number of arguments from the command line
2084  * @param argv command line arguments
2085  * @return 0 ok, 1 on error
2086  */
2087   int
2088 main (int argc, char *const *argv)
2089 {
2090   return (GNUNET_OK ==
2091           GNUNET_SERVICE_run (argc,
2092                               argv,
2093                               "rps",
2094                               GNUNET_SERVICE_OPTION_NONE,
2095                               &run, NULL)) ? 0 : 1;
2096 }
2097
2098 /* end of gnunet-service-rps.c */