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 #ifdef 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   /* Cancle is_live_task if still scheduled */
 619   if (NULL != peer_ctx->is_live_task)
 620   {
 621     GNUNET_CADET_notify_transmit_ready_cancel (peer_ctx->is_live_task);
 622     peer_ctx->is_live_task = NULL;
 623   }
 624
 625   peer = &peer_ctx->peer_id;
 626   set_peer_flag (peer_ctx, VALID);
 627
 628   LOG (GNUNET_ERROR_TYPE_DEBUG, "Peer %s is live\n", GNUNET_i2s (peer));
 629
 630   if (0 != peer_ctx->num_outstanding_ops)
 631   { /* Call outstanding operations */
 632     unsigned int i;
 633
 634     for (i = 0 ; i < peer_ctx->num_outstanding_ops ; i++)
 635       peer_ctx->outstanding_ops[i].op (peer_ctx->outstanding_ops[i].op_cls, peer);
 636     GNUNET_array_grow (peer_ctx->outstanding_ops, peer_ctx->num_outstanding_ops, 0);
 637   }
 638
 639   return 0;
 640 }
 641
 642
 643 /**
 644  * Callback that is called when a channel was effectively established.
 645  * This is given to ntfy_tmt_rdy and called when the channel was
 646  * successfully established.
 647  */
 648 static size_t
 649 cadet_ntfy_tmt_rdy_cb (void *cls, size_t size, void *buf)
 650 {
 651   struct PeerContext *peer_ctx = (struct PeerContext *) cls;
 652
 653   if (NULL != buf
 654       && 0 != size)
 655   {
 656     peer_ctx->is_live_task = NULL;
 657     peer_is_live (peer_ctx);
 658   }
 659
 660   //if (NULL != peer_ctx->is_live_task)
 661   //{
 662   //  LOG (GNUNET_ERROR_TYPE_DEBUG,
 663   //       "Trying to cancle is_live_task for peer %s\n",
 664   //       GNUNET_i2s (&peer_ctx->peer_id));
 665   //  GNUNET_CADET_notify_transmit_ready_cancel (peer_ctx->is_live_task);
 666   //  peer_ctx->is_live_task = NULL;
 667   //}
 668
 669   return 0;
 670 }
 671
 672
 673 /**
 674  * Get the channel of a peer. If not existing, create.
 675  */
 676   struct GNUNET_CADET_Channel *
 677 get_channel (struct GNUNET_CONTAINER_MultiPeerMap *peer_map,
 678              const struct GNUNET_PeerIdentity *peer)
 679 {
 680   struct PeerContext *peer_ctx;
 681
 682   peer_ctx = get_peer_ctx (peer_map, peer);
 683   if (NULL == peer_ctx->send_channel)
 684   {
 685     LOG (GNUNET_ERROR_TYPE_DEBUG,
 686          "Trying to establish channel to peer %s\n",
 687          GNUNET_i2s (peer));
 688
 689     peer_ctx->send_channel =
 690       GNUNET_CADET_channel_create (cadet_handle,
 691                                    NULL,
 692                                    peer,
 693                                    GNUNET_RPS_CADET_PORT,
 694                                    GNUNET_CADET_OPTION_RELIABLE);
 695
 696     // do I have to explicitly put it in the peer_map?
 697     (void) GNUNET_CONTAINER_multipeermap_put
 698       (peer_map,
 699        peer,
 700        peer_ctx,
 701        GNUNET_CONTAINER_MULTIHASHMAPOPTION_REPLACE);
 702   }
 703   return peer_ctx->send_channel;
 704 }
 705
 706
 707 /**
 708  * Get the message queue of a specific peer.
 709  *
 710  * If we already have a message queue open to this client,
 711  * simply return it, otherways create one.
 712  */
 713   struct GNUNET_MQ_Handle *
 714 get_mq (struct GNUNET_CONTAINER_MultiPeerMap *peer_map,
 715         const struct GNUNET_PeerIdentity *peer_id)
 716 {
 717   struct PeerContext *peer_ctx;
 718
 719   peer_ctx = get_peer_ctx (peer_map, peer_id);
 720
 721   if (NULL == peer_ctx->mq)
 722   {
 723     (void) get_channel (peer_map, peer_id);
 724     peer_ctx->mq = GNUNET_CADET_mq_create (peer_ctx->send_channel);
 725     //do I have to explicitly put it in the peer_map?
 726     (void) GNUNET_CONTAINER_multipeermap_put (peer_map, peer_id, peer_ctx,
 727                                        GNUNET_CONTAINER_MULTIHASHMAPOPTION_REPLACE);
 728   }
 729   return peer_ctx->mq;
 730 }
 731
 732
 733 /**
 734  * Issue check whether peer is live
 735  *
 736  * @param peer_ctx the context of the peer
 737  */
 738 void
 739 check_peer_live (struct PeerContext *peer_ctx)
 740 {
 741   (void) get_channel (peer_map, &peer_ctx->peer_id);
 742   LOG (GNUNET_ERROR_TYPE_DEBUG,
 743        "Get informed about peer %s getting live\n",
 744        GNUNET_i2s (&peer_ctx->peer_id));
 745   peer_ctx->is_live_task =
 746       GNUNET_CADET_notify_transmit_ready (peer_ctx->send_channel,
 747                                           GNUNET_NO,
 748                                           GNUNET_TIME_UNIT_FOREVER_REL,
 749                                           sizeof (struct GNUNET_MessageHeader),
 750                                           cadet_ntfy_tmt_rdy_cb,
 751                                           peer_ctx);
 752   // FIXME check whether this is NULL
 753 }
 754
 755
 756 /**
 757  * Sum all time relatives of an array.
 758   */
 759   struct GNUNET_TIME_Relative
 760 T_relative_sum (const struct GNUNET_TIME_Relative *rel_array, uint32_t arr_size)
 761 {
 762   struct GNUNET_TIME_Relative sum;
 763   uint32_t i;
 764
 765   sum = GNUNET_TIME_UNIT_ZERO;
 766   for ( i = 0 ; i < arr_size ; i++ )
 767   {
 768     sum = GNUNET_TIME_relative_add (sum, rel_array[i]);
 769   }
 770   return sum;
 771 }
 772
 773
 774 /**
 775  * Compute the average of given time relatives.
 776  */
 777   struct GNUNET_TIME_Relative
 778 T_relative_avg (const struct GNUNET_TIME_Relative *rel_array, uint32_t arr_size)
 779 {
 780   return GNUNET_TIME_relative_divide (T_relative_sum (rel_array, arr_size), arr_size);
 781 }
 782
 783
 784 /**
 785  * Insert PeerID in #pull_list
 786  *
 787  * Called once we know a peer is live.
 788  */
 789   void
 790 insert_in_pull_list (void *cls, const struct GNUNET_PeerIdentity *peer)
 791 {
 792   if (GNUNET_NO == in_arr (pull_list, pull_list_size, peer))
 793     GNUNET_array_append (pull_list, pull_list_size, *peer);
 794
 795   peer_clean (peer);
 796 }
 797
 798 /**
 799  * Check whether #insert_in_pull_list was already scheduled
 800  */
 801   int
 802 insert_in_pull_list_scheduled (const struct PeerContext *peer_ctx)
 803 {
 804   unsigned int i;
 805
 806   for ( i = 0 ; i < peer_ctx->num_outstanding_ops ; i++ )
 807     if (insert_in_pull_list == peer_ctx->outstanding_ops[i].op)
 808       return GNUNET_YES;
 809   return GNUNET_NO;
 810 }
 811
 812
 813 /**
 814  * Insert PeerID in #gossip_list
 815  *
 816  * Called once we know a peer is live.
 817  */
 818   void
 819 insert_in_gossip_list (void *cls, const struct GNUNET_PeerIdentity *peer)
 820 {
 821   if (GNUNET_NO == in_arr (gossip_list, gossip_list_size, peer))
 822     GNUNET_array_append (gossip_list, gossip_list_size, *peer);
 823
 824   (void) get_channel (peer_map, peer);
 825 }
 826
 827 /**
 828  * Check whether #insert_in_gossip_list was already scheduled
 829  */
 830   int
 831 insert_in_gossip_list_scheduled (const struct PeerContext *peer_ctx)
 832 {
 833   unsigned int i;
 834
 835   for ( i = 0 ; i < peer_ctx->num_outstanding_ops ; i++ )
 836     if (insert_in_gossip_list == peer_ctx->outstanding_ops[i].op)
 837       return GNUNET_YES;
 838   return GNUNET_NO;
 839 }
 840
 841
 842 /**
 843  * Update sampler with given PeerID.
 844  */
 845   void
 846 insert_in_sampler (void *cls, const struct GNUNET_PeerIdentity *peer)
 847 {
 848   LOG (GNUNET_ERROR_TYPE_DEBUG,
 849        "Updating samplers with peer %s from insert_in_sampler()\n",
 850        GNUNET_i2s (peer));
 851   RPS_sampler_update (prot_sampler,   peer);
 852   RPS_sampler_update (client_sampler, peer);
 853 }
 854
 855
 856 /**
 857  * Check whether #insert_in_sampler was already scheduled
 858  */
 859 static int
 860 insert_in_sampler_scheduled (const struct PeerContext *peer_ctx)
 861 {
 862   unsigned int i;
 863
 864   for ( i = 0 ; i < peer_ctx->num_outstanding_ops ; i++ )
 865     if (insert_in_sampler== peer_ctx->outstanding_ops[i].op)
 866       return GNUNET_YES;
 867   return GNUNET_NO;
 868 }
 869
 870
 871 /**
 872  * Wrapper around #RPS_sampler_resize()
 873  *
 874  * If we do not have enough sampler elements, double current sampler size
 875  * If we have more than enough sampler elements, halv current sampler size
 876  */
 877 static void
 878 resize_wrapper (struct RPS_Sampler *sampler, uint32_t new_size)
 879 {
 880   unsigned int sampler_size;
 881
 882   // TODO statistics
 883   // TODO respect the min, max
 884   sampler_size = RPS_sampler_get_size (sampler);
 885   if (sampler_size > new_size * 4)
 886   { /* Shrinking */
 887     RPS_sampler_resize (sampler, sampler_size / 2);
 888   }
 889   else if (sampler_size < new_size)
 890   { /* Growing */
 891     RPS_sampler_resize (sampler, sampler_size * 2);
 892   }
 893   LOG (GNUNET_ERROR_TYPE_DEBUG, "sampler_size is now %u\n", sampler_size);
 894 }
 895
 896
 897 /**
 898  * Wrapper around #RPS_sampler_resize() resizing the client sampler
 899  */
 900 static void
 901 client_resize_wrapper ()
 902 {
 903   uint32_t bigger_size;
 904   unsigned int sampler_size;
 905
 906   // TODO statistics
 907
 908   sampler_size = RPS_sampler_get_size (client_sampler);
 909
 910   if (sampler_size_est_need > sampler_size_client_need)
 911     bigger_size = sampler_size_est_need;
 912   else
 913     bigger_size = sampler_size_client_need;
 914
 915   // TODO respect the min, max
 916   resize_wrapper (client_sampler, bigger_size);
 917   LOG (GNUNET_ERROR_TYPE_DEBUG, "sampler_size is now %u\n", sampler_size);
 918 }
 919
 920
 921 /**
 922  * Estimate request rate
 923  *
 924  * Called every time we receive a request from the client.
 925  */
 926   void
 927 est_request_rate()
 928 {
 929   struct GNUNET_TIME_Relative max_round_duration;
 930
 931   if (request_deltas_size > req_counter)
 932     req_counter++;
 933   if ( 1 < req_counter)
 934   {
 935     /* Shift last request deltas to the right */
 936     memcpy (&request_deltas[1],
 937         request_deltas,
 938         (req_counter - 1) * sizeof (struct GNUNET_TIME_Relative));
 939
 940     /* Add current delta to beginning */
 941     request_deltas[0] =
 942         GNUNET_TIME_absolute_get_difference (last_request,
 943                                              GNUNET_TIME_absolute_get ());
 944     request_rate = T_relative_avg (request_deltas, req_counter);
 945
 946     /* Compute the duration a round will maximally take */
 947     max_round_duration =
 948         GNUNET_TIME_relative_add (round_interval,
 949                                   GNUNET_TIME_relative_divide (round_interval, 2));
 950
 951     /* Set the estimated size the sampler has to have to
 952      * satisfy the current client request rate */
 953     sampler_size_client_need =
 954         max_round_duration.rel_value_us / request_rate.rel_value_us;
 955
 956     /* Resize the sampler */
 957     client_resize_wrapper ();
 958   }
 959   last_request = GNUNET_TIME_absolute_get ();
 960 }
 961
 962
 963 /***********************************************************************
 964  * /Util functions
 965 ***********************************************************************/
 966
 967
 968
 969
 970
 971 /**
 972  * Function called by NSE.
 973  *
 974  * Updates sizes of sampler list and gossip list and adapt those lists
 975  * accordingly.
 976  */
 977   void
 978 nse_callback (void *cls, struct GNUNET_TIME_Absolute timestamp,
 979               double logestimate, double std_dev)
 980 {
 981   double estimate;
 982   //double scale; // TODO this might go gloabal/config
 983
 984   LOG (GNUNET_ERROR_TYPE_DEBUG,
 985        "Received a ns estimate - logest: %f, std_dev: %f (old_size: %u)\n",
 986        logestimate, std_dev, RPS_sampler_get_size (prot_sampler));
 987   //scale = .01;
 988   estimate = GNUNET_NSE_log_estimate_to_n (logestimate);
 989   // GNUNET_NSE_log_estimate_to_n (logestimate);
 990   estimate = pow (estimate, 1.0 / 3);
 991   // TODO add if std_dev is a number
 992   // estimate += (std_dev * scale);
 993   if (2 < ceil (estimate))
 994   {
 995     LOG (GNUNET_ERROR_TYPE_DEBUG, "Changing estimate to %f\n", estimate);
 996     sampler_size_est_need = estimate;
 997   } else
 998     LOG (GNUNET_ERROR_TYPE_DEBUG, "Not using estimate %f\n", estimate);
 999
1000   /* If the NSE has changed adapt the lists accordingly */
1001   resize_wrapper (prot_sampler, sampler_size_est_need);
1002   client_resize_wrapper ();
1003 }
1004
1005
1006 /**
1007  * Callback called once the requested PeerIDs are ready.
1008  *
1009  * Sends those to the requesting client.
1010  */
1011 void client_respond (void *cls,
1012     struct GNUNET_PeerIdentity *ids, uint32_t num_peers)
1013 {
1014   LOG (GNUNET_ERROR_TYPE_DEBUG, "sampler returned %" PRIX32 " peers\n", num_peers);
1015   struct GNUNET_MQ_Envelope *ev;
1016   struct GNUNET_RPS_CS_ReplyMessage *out_msg;
1017   struct GNUNET_SERVER_Client *client;
1018   uint32_t size_needed;
1019   struct client_ctx *cli_ctx;
1020
1021   client = (struct GNUNET_SERVER_Client *) cls;
1022
1023   size_needed = sizeof (struct GNUNET_RPS_CS_ReplyMessage) +
1024                 num_peers * sizeof (struct GNUNET_PeerIdentity);
1025
1026   GNUNET_assert (GNUNET_SERVER_MAX_MESSAGE_SIZE >= size_needed);
1027
1028   ev = GNUNET_MQ_msg_extra (out_msg,
1029                             num_peers * sizeof (struct GNUNET_PeerIdentity),
1030                             GNUNET_MESSAGE_TYPE_RPS_CS_REPLY);
1031   out_msg->num_peers = htonl (num_peers);
1032
1033   memcpy (&out_msg[1],
1034       ids,
1035       num_peers * sizeof (struct GNUNET_PeerIdentity));
1036   GNUNET_free (ids);
1037
1038   cli_ctx = GNUNET_SERVER_client_get_user_context (client, struct client_ctx);
1039   if (NULL == cli_ctx) {
1040     cli_ctx = GNUNET_new (struct client_ctx);
1041     cli_ctx->mq = GNUNET_MQ_queue_for_server_client (client);
1042     GNUNET_SERVER_client_set_user_context (client, cli_ctx);
1043   }
1044
1045   GNUNET_MQ_send (cli_ctx->mq, ev);
1046 }
1047
1048
1049 /**
1050  * Handle RPS request from the client.
1051  *
1052  * @param cls closure
1053  * @param client identification of the client
1054  * @param message the actual message
1055  */
1056 static void
1057 handle_client_request (void *cls,
1058             struct GNUNET_SERVER_Client *client,
1059             const struct GNUNET_MessageHeader *message)
1060 {
1061   struct GNUNET_RPS_CS_RequestMessage *msg;
1062   uint32_t num_peers;
1063   uint32_t size_needed;
1064   uint32_t i;
1065
1066   msg = (struct GNUNET_RPS_CS_RequestMessage *) message;
1067
1068   num_peers = ntohl (msg->num_peers);
1069   size_needed = sizeof (struct GNUNET_RPS_CS_RequestMessage) +
1070                 num_peers * sizeof (struct GNUNET_PeerIdentity);
1071
1072   if (GNUNET_SERVER_MAX_MESSAGE_SIZE < size_needed)
1073   {
1074     GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
1075     return;
1076   }
1077
1078   for (i = 0 ; i < num_peers ; i++)
1079     est_request_rate();
1080
1081   LOG (GNUNET_ERROR_TYPE_DEBUG, "Client requested %" PRIX32 " random peer(s).\n", num_peers);
1082
1083   RPS_sampler_get_n_rand_peers (client_sampler, client_respond,
1084                                 client, num_peers, GNUNET_YES);
1085
1086   GNUNET_SERVER_receive_done (client,
1087                               GNUNET_OK);
1088 }
1089
1090
1091 /**
1092  * Handle seed from the client.
1093  *
1094  * @param cls closure
1095  * @param client identification of the client
1096  * @param message the actual message
1097  */
1098   static void
1099 handle_client_seed (void *cls,
1100                     struct GNUNET_SERVER_Client *client,
1101                     const struct GNUNET_MessageHeader *message)
1102 {
1103   struct GNUNET_RPS_CS_SeedMessage *in_msg;
1104   struct GNUNET_PeerIdentity *peers;
1105   uint32_t num_peers;
1106   uint32_t i;
1107
1108   if (sizeof (struct GNUNET_RPS_CS_SeedMessage) > ntohs (message->size))
1109   {
1110     GNUNET_break_op (0);
1111     GNUNET_SERVER_receive_done (client,
1112                                 GNUNET_SYSERR);
1113   }
1114
1115   in_msg = (struct GNUNET_RPS_CS_SeedMessage *) message;
1116   num_peers = ntohl (in_msg->num_peers);
1117   peers = (struct GNUNET_PeerIdentity *) &in_msg[1];
1118   //peers = GNUNET_new_array (num_peers, struct GNUNET_PeerIdentity);
1119   //memcpy (peers, &in_msg[1], num_peers * sizeof (struct GNUNET_PeerIdentity));
1120
1121   if ((ntohs (message->size) - sizeof (struct GNUNET_RPS_CS_SeedMessage)) /
1122       sizeof (struct GNUNET_PeerIdentity) != num_peers)
1123   {
1124     GNUNET_break_op (0);
1125     GNUNET_SERVER_receive_done (client,
1126                                 GNUNET_SYSERR);
1127   }
1128
1129   LOG (GNUNET_ERROR_TYPE_DEBUG,
1130        "Client seeded peers:\n");
1131   print_peer_list (peers, num_peers);
1132
1133   // TODO check for validity of ids
1134
1135   for (i = 0 ; i < num_peers ; i++)
1136   {
1137     LOG (GNUNET_ERROR_TYPE_DEBUG,
1138          "Updating samplers with seed %" PRIX32 ": %s\n",
1139          i,
1140          GNUNET_i2s (&peers[i]));
1141
1142     RPS_sampler_update (prot_sampler,   &peers[i]);
1143     RPS_sampler_update (client_sampler, &peers[i]);
1144   }
1145
1146   //GNUNET_free (peers);
1147
1148   GNUNET_SERVER_receive_done (client,
1149                                                 GNUNET_OK);
1150 }
1151
1152
1153 /**
1154  * Handle a PUSH message from another peer.
1155  *
1156  * Check the proof of work and store the PeerID
1157  * in the temporary list for pushed PeerIDs.
1158  *
1159  * @param cls Closure
1160  * @param channel The channel the PUSH was received over
1161  * @param channel_ctx The context associated with this channel
1162  * @param msg The message header
1163  */
1164 static int
1165 handle_peer_push (void *cls,
1166     struct GNUNET_CADET_Channel *channel,
1167     void **channel_ctx,
1168     const struct GNUNET_MessageHeader *msg)
1169 {
1170   const struct GNUNET_PeerIdentity *peer;
1171
1172   // (check the proof of work)
1173
1174   peer = (const struct GNUNET_PeerIdentity *) GNUNET_CADET_channel_get_info (channel, GNUNET_CADET_OPTION_PEER);
1175   // FIXME wait for cadet to change this function
1176   LOG (GNUNET_ERROR_TYPE_DEBUG, "PUSH received (%s)\n", GNUNET_i2s (peer));
1177
1178   /* Add the sending peer to the push_list */
1179   if (GNUNET_NO == in_arr (push_list, push_list_size, peer))
1180     GNUNET_array_append (push_list, push_list_size, *peer);
1181
1182   return GNUNET_OK;
1183 }
1184
1185 /**
1186  * Handle PULL REQUEST request message from another peer.
1187  *
1188  * Reply with the gossip list of PeerIDs.
1189  *
1190  * @param cls Closure
1191  * @param channel The channel the PUSH was received over
1192  * @param channel_ctx The context associated with this channel
1193  * @param msg The message header
1194  */
1195 static int
1196 handle_peer_pull_request (void *cls,
1197     struct GNUNET_CADET_Channel *channel,
1198     void **channel_ctx,
1199     const struct GNUNET_MessageHeader *msg)
1200 {
1201   struct GNUNET_PeerIdentity *peer;
1202   uint32_t send_size;
1203   struct GNUNET_MQ_Handle *mq;
1204   struct GNUNET_MQ_Envelope *ev;
1205   struct GNUNET_RPS_P2P_PullReplyMessage *out_msg;
1206
1207
1208   peer = (struct GNUNET_PeerIdentity *)
1209     GNUNET_CADET_channel_get_info (channel,
1210                                    GNUNET_CADET_OPTION_PEER);
1211   // FIXME wait for cadet to change this function
1212
1213   /* Compute actual size */
1214   send_size = sizeof (struct GNUNET_RPS_P2P_PullReplyMessage) +
1215               gossip_list_size * sizeof (struct GNUNET_PeerIdentity);
1216
1217   if (GNUNET_CONSTANTS_MAX_CADET_MESSAGE_SIZE < send_size)
1218     /* Compute number of peers to send
1219      * If too long, simply truncate */
1220   // TODO select random ones via permutation
1221     send_size =
1222       (GNUNET_CONSTANTS_MAX_CADET_MESSAGE_SIZE -
1223        sizeof (struct GNUNET_RPS_P2P_PullReplyMessage)) /
1224        sizeof (struct GNUNET_PeerIdentity);
1225   else
1226     send_size = gossip_list_size;
1227
1228   LOG (GNUNET_ERROR_TYPE_DEBUG,
1229       "PULL REQUEST from peer %s received, going to send %u peers\n",
1230       GNUNET_i2s (peer), send_size);
1231
1232   mq = get_mq (peer_map, peer);
1233
1234   ev = GNUNET_MQ_msg_extra (out_msg,
1235                            send_size * sizeof (struct GNUNET_PeerIdentity),
1236                            GNUNET_MESSAGE_TYPE_RPS_PP_PULL_REPLY);
1237   //out_msg->num_peers = htonl (gossip_list_size);
1238   out_msg->num_peers = htonl (send_size);
1239   memcpy (&out_msg[1], gossip_list,
1240          send_size * sizeof (struct GNUNET_PeerIdentity));
1241
1242   GNUNET_MQ_send (mq, ev);
1243
1244   return GNUNET_OK;
1245 }
1246
1247
1248 /**
1249  * Handle PULL REPLY message from another peer.
1250  *
1251  * Check whether we sent a corresponding request and
1252  * whether this reply is the first one.
1253  *
1254  * @param cls Closure
1255  * @param channel The channel the PUSH was received over
1256  * @param channel_ctx The context associated with this channel
1257  * @param msg The message header
1258  */
1259   static int
1260 handle_peer_pull_reply (void *cls,
1261                         struct GNUNET_CADET_Channel *channel,
1262                         void **channel_ctx,
1263                         const struct GNUNET_MessageHeader *msg)
1264 {
1265   LOG (GNUNET_ERROR_TYPE_DEBUG, "PULL REPLY received\n");
1266
1267   struct GNUNET_RPS_P2P_PullReplyMessage *in_msg;
1268   struct GNUNET_PeerIdentity *peers;
1269   struct PeerContext *peer_ctx;
1270   struct GNUNET_PeerIdentity *sender;
1271   struct PeerContext *sender_ctx;
1272   struct PeerOutstandingOp out_op;
1273   uint32_t i;
1274
1275   /* Check for protocol violation */
1276   if (sizeof (struct GNUNET_RPS_P2P_PullReplyMessage) > ntohs (msg->size))
1277   {
1278     GNUNET_break_op (0);
1279     return GNUNET_SYSERR;
1280   }
1281
1282   in_msg = (struct GNUNET_RPS_P2P_PullReplyMessage *) msg;
1283   if ((ntohs (msg->size) - sizeof (struct GNUNET_RPS_P2P_PullReplyMessage)) /
1284       sizeof (struct GNUNET_PeerIdentity) != ntohl (in_msg->num_peers))
1285   {
1286     LOG (GNUNET_ERROR_TYPE_ERROR,
1287         "message says it sends %" PRIu64 " peers, have space for %i peers\n",
1288         ntohl (in_msg->num_peers),
1289         (ntohs (msg->size) - sizeof (struct GNUNET_RPS_P2P_PullReplyMessage)) /
1290             sizeof (struct GNUNET_PeerIdentity));
1291     GNUNET_break_op (0);
1292     return GNUNET_SYSERR;
1293   }
1294
1295   sender = (struct GNUNET_PeerIdentity *) GNUNET_CADET_channel_get_info (
1296       (struct GNUNET_CADET_Channel *) channel, GNUNET_CADET_OPTION_PEER);
1297        // Guess simply casting isn't the nicest way...
1298        // FIXME wait for cadet to change this function
1299   sender_ctx = get_peer_ctx (peer_map, sender);
1300
1301   if (GNUNET_YES == get_peer_flag (sender_ctx, PULL_REPLY_PENDING))
1302   {
1303     GNUNET_break_op (0);
1304     return GNUNET_OK;
1305   }
1306
1307
1308   /* Do actual logic */
1309   peers = (struct GNUNET_PeerIdentity *) &msg[1];
1310   for (i = 0 ; i < ntohl (in_msg->num_peers) ; i++)
1311   {
1312     peer_ctx = get_peer_ctx (peer_map, &peers[i]);
1313     if (GNUNET_YES == get_peer_flag (peer_ctx, VALID)
1314         || NULL != peer_ctx->send_channel
1315         || NULL != peer_ctx->recv_channel)
1316     {
1317       if (GNUNET_NO == in_arr (pull_list, pull_list_size, &peers[i])
1318           && 0 != GNUNET_CRYPTO_cmp_peer_identity (&own_identity, &peers[i]))
1319         GNUNET_array_append (pull_list, pull_list_size, peers[i]);
1320     }
1321     else if (GNUNET_NO == insert_in_pull_list_scheduled (peer_ctx))
1322     {
1323       out_op.op = insert_in_pull_list;
1324       out_op.op_cls = NULL;
1325       GNUNET_array_append (peer_ctx->outstanding_ops,
1326                            peer_ctx->num_outstanding_ops,
1327                            out_op);
1328       check_peer_live (peer_ctx);
1329     }
1330   }
1331
1332   unset_peer_flag (sender_ctx, PULL_REPLY_PENDING);
1333   rem_from_list (&pending_pull_reply_list, &pending_pull_reply_list_size, sender);
1334
1335   return GNUNET_OK;
1336 }
1337
1338
1339 static void
1340 do_round (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc);
1341
1342 static void
1343 do_mal_round (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc);
1344
1345
1346 #ifdef ENABLE_MALICIOUS
1347 /**
1348  * Turn RPS service to act malicious.
1349  *
1350  * @param cls Closure
1351  * @param channel The channel the PUSH was received over
1352  * @param channel_ctx The context associated with this channel
1353  * @param msg The message header
1354  */
1355   static void
1356 handle_client_act_malicious (void *cls,
1357                              struct GNUNET_SERVER_Client *client,
1358                              const struct GNUNET_MessageHeader *msg)
1359 {
1360   struct GNUNET_RPS_CS_ActMaliciousMessage *in_msg;
1361   struct GNUNET_PeerIdentity *peers;
1362
1363   /* Check for protocol violation */
1364   if (sizeof (struct GNUNET_RPS_CS_ActMaliciousMessage) > ntohs (msg->size))
1365   {
1366     GNUNET_break_op (0);
1367   }
1368
1369   in_msg = (struct GNUNET_RPS_CS_ActMaliciousMessage *) msg;
1370   if ((ntohs (msg->size) - sizeof (struct GNUNET_RPS_CS_ActMaliciousMessage)) /
1371       sizeof (struct GNUNET_PeerIdentity) != ntohl (in_msg->num_peers))
1372   {
1373     LOG (GNUNET_ERROR_TYPE_ERROR,
1374         "message says it sends %" PRIu64 " peers, have space for %i peers\n",
1375         ntohl (in_msg->num_peers),
1376         (ntohs (msg->size) - sizeof (struct GNUNET_RPS_CS_ActMaliciousMessage)) /
1377             sizeof (struct GNUNET_PeerIdentity));
1378     GNUNET_break_op (0);
1379   }
1380
1381
1382   /* Do actual logic */
1383   peers = (struct GNUNET_PeerIdentity *) &msg[1];
1384   num_mal_peers = ntohl (in_msg->num_peers);
1385   mal_type = ntohl (in_msg->type);
1386   num_attacked_peers = 0;
1387   attacked_peers = NULL;
1388
1389   LOG (GNUNET_ERROR_TYPE_DEBUG,
1390        "Now acting malicious type %" PRIu32 "\n",
1391        mal_type);
1392
1393   if (1 == mal_type)
1394   { /* Try to maximise representation */
1395     num_mal_peers = ntohl (in_msg->num_peers);
1396     mal_peers = GNUNET_new_array (num_mal_peers,
1397                                   struct GNUNET_PeerIdentity);
1398     memcpy (mal_peers, peers, num_mal_peers * sizeof (struct GNUNET_PeerIdentity));
1399
1400     /* Substitute do_round () with do_mal_round () */
1401     GNUNET_SCHEDULER_cancel (do_round_task);
1402     do_round_task = GNUNET_SCHEDULER_add_now (&do_mal_round, NULL);
1403   }
1404   else if (2 == mal_type)
1405   { /* Try to partition the network */
1406     num_mal_peers = ntohl (in_msg->num_peers) - 1;
1407     mal_peers = GNUNET_new_array (num_mal_peers,
1408                                   struct GNUNET_PeerIdentity);
1409     memcpy (mal_peers, peers, num_mal_peers);
1410
1411     GNUNET_array_grow (attacked_peers, num_attacked_peers, 1);
1412     memcpy (attacked_peers, &peers[num_mal_peers], 1 * sizeof (struct GNUNET_PeerIdentity));
1413
1414     /* Substitute do_round () with do_mal_round () */
1415     GNUNET_SCHEDULER_cancel (do_round_task);
1416     do_round_task = GNUNET_SCHEDULER_add_now (&do_mal_round, NULL);
1417   }
1418   else if (0 == mal_type)
1419   { /* Stop acting malicious */
1420     num_mal_peers = 0;
1421     GNUNET_free (mal_peers);
1422
1423     /* Substitute do_mal_round () with do_round () */
1424     GNUNET_SCHEDULER_cancel (do_round_task);
1425     do_round_task = GNUNET_SCHEDULER_add_now (&do_round, NULL);
1426   }
1427   else
1428   {
1429     GNUNET_break (0);
1430   }
1431
1432 }
1433
1434
1435 /**
1436  * Send out PUSHes and PULLs maliciously.
1437  *
1438  * This is executed regylary.
1439  */
1440 static void
1441 do_mal_round (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
1442 {
1443   uint32_t num_pushes;
1444   uint32_t i;
1445   unsigned int rand_delay;
1446   struct GNUNET_TIME_Relative half_round_interval;
1447   struct GNUNET_TIME_Relative time_next_round;
1448
1449   LOG (GNUNET_ERROR_TYPE_DEBUG, "Going to execute next round maliciously.\n");
1450
1451   /* Do malicious actions */
1452   if (1 == mal_type)
1453   { /* Try to maximise representation */
1454     num_pushes = min (min (push_limit, /* FIXME: attacked peer */ num_mal_peers), GNUNET_CONSTANTS_MAX_CADET_MESSAGE_SIZE);
1455     for (i = 0 ; i < num_pushes ; i++)
1456     { /* Send PUSH to attacked peer */
1457       //GNUNET_CONTAINER_multihashmap_iterator_create
1458     }
1459   }
1460   else if (2 == mal_type)
1461   { /* Try to partition the network */
1462     /* Send as many pushes to attacked peer as possible */
1463   }
1464
1465   /* Compute random time value between .5 * round_interval and 1.5 *round_interval */
1466   half_round_interval = GNUNET_TIME_relative_divide (round_interval, 2);
1467   do
1468   {
1469   /*
1470    * Compute random value between (0 and 1) * round_interval
1471    * via multiplying round_interval with a 'fraction' (0 to value)/value
1472    */
1473   rand_delay = GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK, UINT_MAX/10);
1474   time_next_round = GNUNET_TIME_relative_multiply (round_interval,  rand_delay);
1475   time_next_round = GNUNET_TIME_relative_divide   (time_next_round, UINT_MAX/10);
1476   time_next_round = GNUNET_TIME_relative_add      (time_next_round, half_round_interval);
1477   } while (GNUNET_TIME_UNIT_FOREVER_REL.rel_value_us == time_next_round.rel_value_us);
1478
1479   /* Schedule next round */
1480   do_round_task = GNUNET_SCHEDULER_add_delayed (round_interval, &do_mal_round, NULL);
1481   LOG (GNUNET_ERROR_TYPE_DEBUG, "Finished round\n");
1482 }
1483 #endif /* ENABLE_MALICIOUS */
1484
1485
1486 /**
1487  * Send out PUSHes and PULLs, possibly update #gossip_list, samplers.
1488  *
1489  * This is executed regylary.
1490  */
1491 static void
1492 do_round (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
1493 {
1494   LOG (GNUNET_ERROR_TYPE_DEBUG, "Going to execute next round.\n");
1495
1496   uint32_t i;
1497   unsigned int *permut;
1498   unsigned int n_peers; /* Number of peers we send pushes/pulls to */
1499   struct GNUNET_MQ_Envelope *ev;
1500   struct GNUNET_PeerIdentity peer;
1501   struct GNUNET_PeerIdentity *tmp_peer;
1502   struct GNUNET_MQ_Handle *mq;
1503
1504   LOG (GNUNET_ERROR_TYPE_DEBUG,
1505        "Printing gossip list:\n");
1506   for (i = 0 ; i < gossip_list_size ; i++)
1507     LOG (GNUNET_ERROR_TYPE_DEBUG,
1508          "\t%s\n", GNUNET_i2s (&gossip_list[i]));
1509   // TODO log lists, ...
1510
1511   /* Would it make sense to have one shuffeled gossip list and then
1512    * to send PUSHes to first alpha peers, PULL requests to next beta peers and
1513    * use the rest to update sampler?
1514    * in essence get random peers with consumption */
1515
1516   /* Send PUSHes */
1517   if (0 < gossip_list_size)
1518   {
1519     permut = GNUNET_CRYPTO_random_permute (GNUNET_CRYPTO_QUALITY_STRONG,
1520                                            (unsigned int) gossip_list_size);
1521     n_peers = ceil (alpha * gossip_list_size);
1522     LOG (GNUNET_ERROR_TYPE_DEBUG,
1523          "Going to send pushes to %u ceil (%f * %u) peers.\n",
1524          n_peers, alpha, gossip_list_size);
1525     for (i = 0 ; i < n_peers ; i++)
1526     {
1527       peer = gossip_list[permut[i]];
1528       if (0 != GNUNET_CRYPTO_cmp_peer_identity (&own_identity, &peer)) // TODO
1529       { // FIXME if this fails schedule/loop this for later
1530         LOG (GNUNET_ERROR_TYPE_DEBUG,
1531              "Sending PUSH to peer %s of gossiped list.\n",
1532              GNUNET_i2s (&peer));
1533
1534         ev = GNUNET_MQ_msg_header (GNUNET_MESSAGE_TYPE_RPS_PP_PUSH);
1535         mq = get_mq (peer_map, &peer);
1536         GNUNET_MQ_send (mq, ev);
1537       }
1538     }
1539     GNUNET_free (permut);
1540   }
1541
1542
1543   /* Send PULL requests */
1544   //permut = GNUNET_CRYPTO_random_permute (GNUNET_CRYPTO_QUALITY_STRONG, (unsigned int) sampler_list->size);
1545   n_peers = ceil (beta * gossip_list_size);
1546   LOG (GNUNET_ERROR_TYPE_DEBUG,
1547        "Going to send pulls to %u ceil (%f * %u) peers.\n",
1548        n_peers, beta, gossip_list_size);
1549   for (i = 0 ; i < n_peers ; i++)
1550   {
1551     tmp_peer = get_rand_peer_ignore_list (gossip_list, gossip_list_size,
1552         pending_pull_reply_list, pending_pull_reply_list_size);
1553     if (NULL != tmp_peer)
1554     {
1555       peer = *tmp_peer;
1556       GNUNET_free (tmp_peer);
1557
1558       GNUNET_array_append (pending_pull_reply_list, pending_pull_reply_list_size, peer);
1559
1560       if (0 != GNUNET_CRYPTO_cmp_peer_identity (&own_identity, &peer))
1561       { // FIXME if this fails schedule/loop this for later
1562         LOG (GNUNET_ERROR_TYPE_DEBUG,
1563              "Sending PULL request to peer %s of gossiped list.\n",
1564              GNUNET_i2s (&peer));
1565
1566         ev = GNUNET_MQ_msg_header (GNUNET_MESSAGE_TYPE_RPS_PP_PULL_REQUEST);
1567         mq = get_mq (peer_map, &peer);
1568         GNUNET_MQ_send (mq, ev);
1569       }
1570     }
1571   }
1572
1573
1574   /* Update gossip list */
1575
1576   if ( push_list_size <= alpha * gossip_list_size &&
1577        push_list_size != 0 &&
1578        pull_list_size != 0 )
1579   {
1580     LOG (GNUNET_ERROR_TYPE_DEBUG, "Update of the gossip list.\n");
1581
1582     uint32_t first_border;
1583     uint32_t second_border;
1584     uint32_t r_index;
1585     uint32_t peers_to_clean_size;
1586     struct GNUNET_PeerIdentity *peers_to_clean;
1587
1588     peers_to_clean = NULL;
1589     peers_to_clean_size = 0;
1590     GNUNET_array_grow (peers_to_clean, peers_to_clean_size, gossip_list_size);
1591     memcpy (peers_to_clean,
1592             gossip_list,
1593             gossip_list_size * sizeof (struct GNUNET_PeerIdentity));
1594
1595     first_border  =                ceil (alpha * sampler_size_est_need);
1596     second_border = first_border + ceil (beta  * sampler_size_est_need);
1597
1598     GNUNET_array_grow (gossip_list, gossip_list_size, second_border);
1599
1600     for (i = 0 ; i < first_border ; i++)
1601     { // TODO use RPS_sampler_get_n_rand_peers
1602       /* Update gossip list with peers received through PUSHes */
1603       r_index = GNUNET_CRYPTO_random_u64 (GNUNET_CRYPTO_QUALITY_STRONG,
1604                                        push_list_size);
1605       gossip_list[i] = push_list[r_index];
1606       // TODO change the peer_flags accordingly
1607     }
1608
1609     for (i = first_border ; i < second_border ; i++)
1610     {
1611       /* Update gossip list with peers received through PULLs */
1612       r_index = GNUNET_CRYPTO_random_u64 (GNUNET_CRYPTO_QUALITY_STRONG,
1613                                        pull_list_size);
1614       gossip_list[i] = pull_list[r_index];
1615       // TODO change the peer_flags accordingly
1616     }
1617
1618     for (i = second_border ; i < sampler_size_est_need ; i++)
1619     {
1620       /* Update gossip list with peers from history */
1621       RPS_sampler_get_n_rand_peers (prot_sampler, hist_update, NULL, 1, GNUNET_NO);
1622       num_hist_update_tasks++;
1623       // TODO change the peer_flags accordingly
1624     }
1625
1626     for (i = 0 ; i < gossip_list_size ; i++)
1627       rem_from_list (&peers_to_clean, &peers_to_clean_size, &gossip_list[i]);
1628
1629     for (i = 0 ; i < peers_to_clean_size ; i++)
1630       peer_clean (&peers_to_clean[i]);
1631
1632     GNUNET_free (peers_to_clean);
1633   }
1634   else
1635   {
1636     LOG (GNUNET_ERROR_TYPE_DEBUG, "No update of the gossip list.\n");
1637   }
1638   // TODO independent of that also get some peers from CADET_get_peers()?
1639
1640
1641   /* Update samplers */
1642   for ( i = 0 ; i < push_list_size ; i++ )
1643   {
1644     LOG (GNUNET_ERROR_TYPE_DEBUG,
1645          "Updating with peer %s from push list\n",
1646          GNUNET_i2s (&push_list[i]));
1647     RPS_sampler_update (prot_sampler,   &push_list[i]);
1648     RPS_sampler_update (client_sampler, &push_list[i]);
1649     // TODO set in_flag?
1650   }
1651
1652   for ( i = 0 ; i < pull_list_size ; i++ )
1653   {
1654     LOG (GNUNET_ERROR_TYPE_DEBUG,
1655          "Updating with peer %s from pull list\n",
1656          GNUNET_i2s (&pull_list[i]));
1657     RPS_sampler_update (prot_sampler,   &push_list[i]);
1658     RPS_sampler_update (client_sampler, &push_list[i]);
1659     // TODO set in_flag?
1660   }
1661
1662
1663   /* Empty push/pull lists */
1664   GNUNET_array_grow (push_list, push_list_size, 0);
1665   GNUNET_array_grow (pull_list, pull_list_size, 0);
1666
1667   struct GNUNET_TIME_Relative time_next_round;
1668   struct GNUNET_TIME_Relative half_round_interval;
1669   unsigned int rand_delay;
1670
1671
1672   /* Compute random time value between .5 * round_interval and 1.5 *round_interval */
1673   half_round_interval = GNUNET_TIME_relative_divide (round_interval, 2);
1674   do
1675   {
1676   /*
1677    * Compute random value between (0 and 1) * round_interval
1678    * via multiplying round_interval with a 'fraction' (0 to value)/value
1679    */
1680   rand_delay = GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK, UINT_MAX/10);
1681   time_next_round = GNUNET_TIME_relative_multiply (round_interval,  rand_delay);
1682   time_next_round = GNUNET_TIME_relative_divide   (time_next_round, UINT_MAX/10);
1683   time_next_round = GNUNET_TIME_relative_add      (time_next_round, half_round_interval);
1684   } while (GNUNET_TIME_UNIT_FOREVER_REL.rel_value_us == time_next_round.rel_value_us);
1685
1686   /* Schedule next round */
1687   do_round_task = GNUNET_SCHEDULER_add_delayed (round_interval, &do_round, NULL);
1688   LOG (GNUNET_ERROR_TYPE_DEBUG, "Finished round\n");
1689 }
1690
1691
1692 static void
1693 rps_start (struct GNUNET_SERVER_Handle *server);
1694
1695
1696 /**
1697  * This is called from GNUNET_CADET_get_peers().
1698  *
1699  * It is called on every peer(ID) that cadet somehow has contact with.
1700  * We use those to initialise the sampler.
1701  */
1702 void
1703 init_peer_cb (void *cls,
1704               const struct GNUNET_PeerIdentity *peer,
1705               int tunnel, // "Do we have a tunnel towards this peer?"
1706               unsigned int n_paths, // "Number of known paths towards this peer"
1707               unsigned int best_path) // "How long is the best path?
1708                                       // (0 = unknown, 1 = ourselves, 2 = neighbor)"
1709 {
1710   struct PeerOutstandingOp out_op;
1711   struct PeerContext *peer_ctx;
1712
1713   if (NULL != peer
1714       && 0 != GNUNET_CRYPTO_cmp_peer_identity (&own_identity, peer))
1715   {
1716     LOG (GNUNET_ERROR_TYPE_DEBUG,
1717         "Got peer %s (at %p) from CADET (gossip_list_size: %u)\n",
1718         GNUNET_i2s (peer), peer, gossip_list_size);
1719
1720     // maybe create a function for that
1721     peer_ctx = get_peer_ctx (peer_map, peer);
1722     if (GNUNET_YES != get_peer_flag (peer_ctx, VALID))
1723     {
1724       if (GNUNET_NO == insert_in_sampler_scheduled (peer_ctx))
1725       {
1726         out_op.op = insert_in_sampler;
1727         out_op.op_cls = NULL;
1728         GNUNET_array_append (peer_ctx->outstanding_ops,
1729                              peer_ctx->num_outstanding_ops,
1730                              out_op);
1731       }
1732
1733       if (GNUNET_NO == insert_in_gossip_list_scheduled (peer_ctx))
1734       {
1735         out_op.op = insert_in_gossip_list;
1736         out_op.op_cls = NULL;
1737         GNUNET_array_append (peer_ctx->outstanding_ops,
1738                              peer_ctx->num_outstanding_ops,
1739                              out_op);
1740       }
1741
1742       /* Trigger livelyness test on peer */
1743       check_peer_live (peer_ctx);
1744     }
1745
1746     // send push/pull to each of those peers?
1747   }
1748 }
1749
1750
1751 /**
1752  * Clean the send channel of a peer
1753  */
1754 void
1755 peer_clean (const struct GNUNET_PeerIdentity *peer)
1756 {
1757   struct PeerContext *peer_ctx;
1758   struct GNUNET_CADET_Channel *channel;
1759
1760   if (GNUNET_YES != in_arr (gossip_list, gossip_list_size, peer)
1761       && GNUNET_YES == GNUNET_CONTAINER_multipeermap_contains (peer_map, peer))
1762   {
1763     peer_ctx = get_peer_ctx (peer_map, peer);
1764     if (NULL != peer_ctx->send_channel)
1765     {
1766       channel = peer_ctx->send_channel;
1767       peer_ctx->send_channel = NULL;
1768       GNUNET_CADET_channel_destroy (channel);
1769     }
1770   }
1771 }
1772
1773
1774 /**
1775  * Callback used to remove peers from the multipeermap.
1776  */
1777   int
1778 peer_remove_cb (void *cls, const struct GNUNET_PeerIdentity *key, void *value)
1779 {
1780   struct PeerContext *peer_ctx;
1781   const struct GNUNET_CADET_Channel *channel =
1782     (const struct GNUNET_CADET_Channel *) cls;
1783   struct GNUNET_CADET_Channel *recv;
1784   struct GNUNET_CADET_Channel *send;
1785
1786   if (GNUNET_YES == GNUNET_CONTAINER_multipeermap_contains (peer_map, value))
1787   {
1788     peer_ctx = (struct PeerContext *) value;
1789
1790     if (0 != peer_ctx->num_outstanding_ops)
1791       GNUNET_array_grow (peer_ctx->outstanding_ops,
1792                          peer_ctx->num_outstanding_ops,
1793                          0);
1794
1795     if (NULL != peer_ctx->mq)
1796     {
1797       GNUNET_MQ_destroy (peer_ctx->mq);
1798       peer_ctx->mq = NULL;
1799     }
1800
1801
1802     if (NULL != peer_ctx->is_live_task)
1803     {
1804     LOG (GNUNET_ERROR_TYPE_DEBUG,
1805          "Trying to cancle is_live_task for peer %s\n",
1806          GNUNET_i2s (key));
1807       GNUNET_CADET_notify_transmit_ready_cancel (peer_ctx->is_live_task);
1808       peer_ctx->is_live_task = NULL;
1809     }
1810
1811     send = peer_ctx->send_channel;
1812     peer_ctx->send_channel = NULL;
1813     if (NULL != send
1814         && channel != send)
1815     {
1816       GNUNET_CADET_channel_destroy (send);
1817     }
1818
1819     recv = peer_ctx->send_channel;
1820     peer_ctx->recv_channel = NULL;
1821     if (NULL != recv
1822         && channel != recv)
1823     {
1824       GNUNET_CADET_channel_destroy (recv);
1825     }
1826
1827     if (GNUNET_YES != GNUNET_CONTAINER_multipeermap_remove_all (peer_map, key))
1828       LOG (GNUNET_ERROR_TYPE_WARNING, "removing peer from peer_map failed\n");
1829     else
1830       GNUNET_free (peer_ctx);
1831   }
1832
1833   return GNUNET_YES;
1834 }
1835
1836
1837 /**
1838  * Task run during shutdown.
1839  *
1840  * @param cls unused
1841  * @param tc unused
1842  */
1843 static void
1844 shutdown_task (void *cls,
1845                      const struct GNUNET_SCHEDULER_TaskContext *tc)
1846 {
1847   LOG (GNUNET_ERROR_TYPE_DEBUG, "RPS is going down\n");
1848
1849   if (NULL != do_round_task)
1850   {
1851     GNUNET_SCHEDULER_cancel (do_round_task);
1852     do_round_task = NULL;
1853   }
1854
1855
1856   {
1857   if (GNUNET_SYSERR ==
1858         GNUNET_CONTAINER_multipeermap_iterate (peer_map, peer_remove_cb, NULL))
1859     LOG (GNUNET_ERROR_TYPE_WARNING,
1860         "Iterating over peers to disconnect from them was cancelled\n");
1861   }
1862
1863   GNUNET_NSE_disconnect (nse);
1864   GNUNET_CADET_disconnect (cadet_handle);
1865   RPS_sampler_destroy (prot_sampler);
1866   RPS_sampler_destroy (client_sampler);
1867   LOG (GNUNET_ERROR_TYPE_DEBUG,
1868        "Size of the peermap: %u\n",
1869        GNUNET_CONTAINER_multipeermap_size (peer_map));
1870   GNUNET_break (0 == GNUNET_CONTAINER_multipeermap_size (peer_map));
1871   GNUNET_CONTAINER_multipeermap_destroy (peer_map);
1872   GNUNET_array_grow (gossip_list, gossip_list_size, 0);
1873   GNUNET_array_grow (push_list, push_list_size, 0);
1874   GNUNET_array_grow (pull_list, pull_list_size, 0);
1875 }
1876
1877
1878 /**
1879  * A client disconnected.  Remove all of its data structure entries.
1880  *
1881  * @param cls closure, NULL
1882  * @param client identification of the client
1883  */
1884 static void
1885 handle_client_disconnect (void *cls,
1886                           struct GNUNET_SERVER_Client * client)
1887 {
1888 }
1889
1890
1891 /**
1892  * Handle the channel a peer opens to us.
1893  *
1894  * @param cls The closure
1895  * @param channel The channel the peer wants to establish
1896  * @param initiator The peer's peer ID
1897  * @param port The port the channel is being established over
1898  * @param options Further options
1899  */
1900   static void *
1901 handle_inbound_channel (void *cls,
1902                         struct GNUNET_CADET_Channel *channel,
1903                         const struct GNUNET_PeerIdentity *initiator,
1904                         uint32_t port,
1905                         enum GNUNET_CADET_ChannelOption options)
1906 {
1907   struct PeerContext *peer_ctx;
1908   struct GNUNET_PeerIdentity peer;
1909
1910   peer = *initiator;
1911   LOG (GNUNET_ERROR_TYPE_DEBUG,
1912       "New channel was established to us (Peer %s).\n",
1913       GNUNET_i2s (&peer));
1914
1915   GNUNET_assert (NULL != channel);
1916
1917   // we might not even store the recv_channel
1918
1919   peer_ctx = get_peer_ctx (peer_map, &peer);
1920   // FIXME what do we do if a channel is established twice?
1921   //       overwrite? Clean old channel? ...?
1922   //if (NULL != peer_ctx->recv_channel)
1923   //{
1924   //  peer_ctx->recv_channel = channel;
1925   //}
1926   peer_ctx->recv_channel = channel;
1927
1928   peer_ctx->mq = NULL;
1929
1930   (void) GNUNET_CONTAINER_multipeermap_put (peer_map, &peer, peer_ctx,
1931       GNUNET_CONTAINER_MULTIHASHMAPOPTION_REPLACE);
1932
1933   peer_is_live (peer_ctx);
1934
1935   return NULL; // TODO
1936 }
1937
1938
1939 /**
1940  * This is called when a remote peer destroys a channel.
1941  *
1942  * @param cls The closure
1943  * @param channel The channel being closed
1944  * @param channel_ctx The context associated with this channel
1945  */
1946   static void
1947 cleanup_channel (void *cls,
1948                 const struct GNUNET_CADET_Channel *channel,
1949                 void *channel_ctx)
1950 {
1951   struct GNUNET_PeerIdentity *peer;
1952   struct PeerContext *peer_ctx;
1953
1954   peer = (struct GNUNET_PeerIdentity *) GNUNET_CADET_channel_get_info (
1955       (struct GNUNET_CADET_Channel *) channel, GNUNET_CADET_OPTION_PEER);
1956        // Guess simply casting isn't the nicest way...
1957        // FIXME wait for cadet to change this function
1958   LOG (GNUNET_ERROR_TYPE_DEBUG, "Cleaning up channel to peer %s\n",
1959        GNUNET_i2s (peer));
1960
1961   if (GNUNET_YES == GNUNET_CONTAINER_multipeermap_contains (peer_map, peer))
1962   {
1963     peer_ctx = GNUNET_CONTAINER_multipeermap_get (peer_map, peer);
1964
1965     if (NULL == peer_ctx) /* It could have been removed by shutdown_task */
1966       return;
1967
1968     if (channel == peer_ctx->send_channel)
1969     { /* Peer probably went down */
1970       rem_from_list (&gossip_list, &gossip_list_size, peer);
1971       rem_from_list (&pending_pull_reply_list, &pending_pull_reply_list_size, peer);
1972
1973       /* Somwewhat {ab,re}use the iterator function */
1974       /* Cast to void is ok, because it's used as void in peer_remove_cb */
1975       (void) peer_remove_cb ((void *) channel, peer, peer_ctx);
1976     }
1977     else /* Other peer doesn't want to send us messages anymore */
1978       peer_ctx->recv_channel = NULL;
1979   }
1980 }
1981
1982
1983 /**
1984  * Actually start the service.
1985  */
1986   static void
1987 rps_start (struct GNUNET_SERVER_Handle *server)
1988 {
1989   static const struct GNUNET_SERVER_MessageHandler handlers[] = {
1990     {&handle_client_request,     NULL, GNUNET_MESSAGE_TYPE_RPS_CS_REQUEST,
1991       sizeof (struct GNUNET_RPS_CS_RequestMessage)},
1992     {&handle_client_seed,        NULL, GNUNET_MESSAGE_TYPE_RPS_CS_SEED, 0},
1993     #ifdef ENABLE_MALICIOUS
1994     {&handle_client_act_malicious, NULL, GNUNET_MESSAGE_TYPE_RPS_ACT_MALICIOUS , 0},
1995     #endif /* ENABLE_MALICIOUS */
1996     {NULL, NULL, 0, 0}
1997   };
1998
1999   GNUNET_SERVER_add_handlers (server, handlers);
2000   GNUNET_SERVER_disconnect_notify (server,
2001                                    &handle_client_disconnect,
2002                                    NULL);
2003   LOG (GNUNET_ERROR_TYPE_DEBUG, "Ready to receive requests from clients\n");
2004
2005
2006   do_round_task = GNUNET_SCHEDULER_add_now (&do_round, NULL);
2007   LOG (GNUNET_ERROR_TYPE_DEBUG, "Scheduled first round\n");
2008
2009   GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_FOREVER_REL,
2010                                                         &shutdown_task,
2011                                                         NULL);
2012 }
2013
2014
2015 /**
2016  * Process statistics requests.
2017  *
2018  * @param cls closure
2019  * @param server the initialized server
2020  * @param c configuration to use
2021  */
2022   static void
2023 run (void *cls,
2024      struct GNUNET_SERVER_Handle *server,
2025      const struct GNUNET_CONFIGURATION_Handle *c)
2026 {
2027   // TODO check what this does -- copied from gnunet-boss
2028   // - seems to work as expected
2029   GNUNET_log_setup ("rps", GNUNET_error_type_to_string (GNUNET_ERROR_TYPE_DEBUG), NULL);
2030   cfg = c;
2031
2032
2033   /* Get own ID */
2034   GNUNET_CRYPTO_get_peer_identity (cfg, &own_identity); // TODO check return value
2035   GNUNET_log (GNUNET_ERROR_TYPE_INFO,
2036               "STARTING SERVICE (rps) for peer [%s]\n",
2037               GNUNET_i2s (&own_identity));
2038   #ifdef ENABLE_MALICIOUS
2039   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2040               "Malicious execution compiled in.\n");
2041   #endif /* ENABLE_MALICIOUS */
2042
2043
2044
2045   /* Get time interval from the configuration */
2046   if (GNUNET_OK != GNUNET_CONFIGURATION_get_value_time (cfg, "RPS",
2047                                                         "ROUNDINTERVAL",
2048                                                         &round_interval))
2049   {
2050     LOG (GNUNET_ERROR_TYPE_DEBUG, "Failed to read ROUNDINTERVAL from config\n");
2051     GNUNET_SCHEDULER_shutdown ();
2052     return;
2053   }
2054
2055   /* Get initial size of sampler/gossip list from the configuration */
2056   if (GNUNET_OK != GNUNET_CONFIGURATION_get_value_number (cfg, "RPS",
2057                                                          "INITSIZE",
2058                                                          (long long unsigned int *) &sampler_size_est_need))
2059   {
2060     LOG (GNUNET_ERROR_TYPE_DEBUG, "Failed to read INITSIZE from config\n");
2061     GNUNET_SCHEDULER_shutdown ();
2062     return;
2063   }
2064   LOG (GNUNET_ERROR_TYPE_DEBUG, "INITSIZE is %" PRIu64 "\n", sampler_size_est_need);
2065
2066
2067   gossip_list = NULL;
2068
2069
2070   /* connect to NSE */
2071   nse = GNUNET_NSE_connect (cfg, nse_callback, NULL);
2072   // TODO check whether that was successful
2073   LOG (GNUNET_ERROR_TYPE_DEBUG, "Connected to NSE\n");
2074
2075
2076   alpha = 0.45;
2077   beta  = 0.45;
2078
2079   peer_map = GNUNET_CONTAINER_multipeermap_create (sampler_size_est_need, GNUNET_NO);
2080
2081
2082   /* Initialise cadet */
2083   static const struct GNUNET_CADET_MessageHandler cadet_handlers[] = {
2084     {&handle_peer_push        , GNUNET_MESSAGE_TYPE_RPS_PP_PUSH        ,
2085       sizeof (struct GNUNET_MessageHeader)},
2086     {&handle_peer_pull_request, GNUNET_MESSAGE_TYPE_RPS_PP_PULL_REQUEST,
2087       sizeof (struct GNUNET_MessageHeader)},
2088     {&handle_peer_pull_reply  , GNUNET_MESSAGE_TYPE_RPS_PP_PULL_REPLY  , 0},
2089     {NULL, 0, 0}
2090   };
2091
2092   const uint32_t ports[] = {GNUNET_RPS_CADET_PORT, 0}; // _PORT specified in src/rps/rps.h
2093   cadet_handle = GNUNET_CADET_connect (cfg,
2094                                        cls,
2095                                        &handle_inbound_channel,
2096                                        &cleanup_channel,
2097                                        cadet_handlers,
2098                                        ports);
2099   LOG (GNUNET_ERROR_TYPE_DEBUG, "Connected to CADET\n");
2100
2101
2102   /* Initialise sampler */
2103   struct GNUNET_TIME_Relative half_round_interval;
2104   struct GNUNET_TIME_Relative  max_round_interval;
2105
2106   half_round_interval = GNUNET_TIME_relative_multiply (round_interval, .5);
2107   max_round_interval = GNUNET_TIME_relative_add (round_interval, half_round_interval);
2108
2109   prot_sampler =   RPS_sampler_init (sampler_size_est_need, max_round_interval);
2110   client_sampler = RPS_sampler_init (sampler_size_est_need, max_round_interval);
2111
2112   /* Initialise push and pull maps */
2113   push_list = NULL;
2114   push_list_size = 0;
2115   pull_list = NULL;
2116   pull_list_size = 0;
2117   pending_pull_reply_list = NULL;
2118   pending_pull_reply_list_size = 0;
2119
2120
2121   num_hist_update_tasks = 0;
2122
2123
2124   LOG (GNUNET_ERROR_TYPE_DEBUG, "Requesting peers from CADET\n");
2125   GNUNET_CADET_get_peers (cadet_handle, &init_peer_cb, NULL);
2126   // TODO send push/pull to each of those peers?
2127
2128
2129   rps_start (server);
2130 }
2131
2132
2133 /**
2134  * The main function for the rps service.
2135  *
2136  * @param argc number of arguments from the command line
2137  * @param argv command line arguments
2138  * @return 0 ok, 1 on error
2139  */
2140   int
2141 main (int argc, char *const *argv)
2142 {
2143   return (GNUNET_OK ==
2144           GNUNET_SERVICE_run (argc,
2145                               argv,
2146                               "rps",
2147                               GNUNET_SERVICE_OPTION_NONE,
2148                               &run, NULL)) ? 0 : 1;
2149 }
2150
2151 /* end of gnunet-service-rps.c */