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_peerinfo_service.h"
  30 #include "gnunet_nse_service.h"
  31 #include "rps.h"
  32 #include "rps-test_util.h"
  33
  34 #include "gnunet-service-rps_sampler.h"
  35
  36 #include <math.h>
  37 #include <inttypes.h>
  38
  39 #define LOG(kind, ...) GNUNET_log(kind, __VA_ARGS__)
  40
  41 // TODO modify @brief in every file
  42
  43 // TODO check for overflows
  44
  45 // TODO align message structs
  46
  47 // TODO connect to friends
  48
  49 // TODO store peers somewhere persistent
  50
  51 // TODO blacklist? (-> mal peer detection on top of brahms)
  52
  53 // TODO API request_cancel
  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 ClientContext
  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    * We set this bit when we are going to destroy the channel to this peer.
 105    * When cleanup_channel is called, we know that we wanted to destroy it.
 106    * Otherwise the channel to the other peer was destroyed.
 107    */
 108   TO_DESTROY           = 0x20,
 109 };
 110
 111
 112 /**
 113  * Functions of this type can be used to be stored at a peer for later execution.
 114  */
 115 typedef void (* PeerOp) (void *cls, const struct GNUNET_PeerIdentity *peer);
 116
 117 /**
 118  * Outstanding operation on peer consisting of callback and closure
 119  */
 120 struct PeerOutstandingOp
 121 {
 122   /**
 123    * Callback
 124    */
 125   PeerOp op;
 126
 127   /**
 128    * Closure
 129    */
 130   void *op_cls;
 131 };
 132
 133
 134 /**
 135  * Struct used to keep track of other peer's status
 136  *
 137  * This is stored in a multipeermap.
 138  */
 139 struct PeerContext
 140 {
 141   /**
 142    * In own gossip/sampler list, in other's gossip/sampler list
 143    */
 144   uint32_t peer_flags;
 145
 146   /**
 147    * Message queue open to client
 148    */
 149   struct GNUNET_MQ_Handle *mq;
 150
 151   /**
 152    * Channel open to client.
 153    */
 154   struct GNUNET_CADET_Channel *send_channel;
 155
 156   /**
 157    * Channel open from client.
 158    */
 159   struct GNUNET_CADET_Channel *recv_channel; // unneeded?
 160
 161   /**
 162    * Array of outstanding operations on this peer.
 163    */
 164   struct PeerOutstandingOp *outstanding_ops;
 165
 166   /**
 167    * Number of outstanding operations.
 168    */
 169   unsigned int num_outstanding_ops;
 170   //size_t num_outstanding_ops;
 171
 172   /**
 173    * Handle to the callback given to cadet_ntfy_tmt_rdy()
 174    *
 175    * To be canceled on shutdown.
 176    */
 177   struct GNUNET_CADET_TransmitHandle *is_live_task;
 178
 179   /**
 180    * Identity of the peer
 181    */
 182   struct GNUNET_PeerIdentity peer_id;
 183
 184   /**
 185    * This is pobably followed by 'statistical' data (when we first saw
 186    * him, how did we get his ID, how many pushes (in a timeinterval),
 187    * ...)
 188    */
 189 };
 190
 191 /***********************************************************************
 192  * /Housekeeping with peers
 193 ***********************************************************************/
 194
 195
 196
 197
 198
 199 /***********************************************************************
 200  * Globals
 201 ***********************************************************************/
 202
 203 /**
 204  * Sampler used for the Brahms protocol itself.
 205  */
 206 static struct RPS_Sampler *prot_sampler;
 207
 208 /**
 209  * Sampler used for the clients.
 210  */
 211 static struct RPS_Sampler *client_sampler;
 212
 213 /**
 214  * Set of all peers to keep track of them.
 215  */
 216 static struct GNUNET_CONTAINER_MultiPeerMap *peer_map;
 217
 218
 219 /**
 220  * The gossiped list of peers.
 221  */
 222 static struct GNUNET_PeerIdentity *gossip_list;
 223
 224 /**
 225  * Size of the gossiped list
 226  */
 227 //static unsigned int gossip_list_size;
 228 static uint32_t gossip_list_size;
 229
 230 /**
 231  * Name to log view (gossip_list) to
 232  */
 233 static char *file_name_view_log;
 234
 235
 236 /**
 237  * The size of sampler we need to be able to satisfy the client's need of
 238  * random peers.
 239  */
 240 static unsigned int sampler_size_client_need;
 241
 242 /**
 243  * The size of sampler we need to be able to satisfy the Brahms protocol's
 244  * need of random peers.
 245  *
 246  * This is directly taken as the #gossip_list_size on update of the
 247  * #gossip_list
 248  *
 249  * This is one minimum size the sampler grows to.
 250  */
 251 static unsigned int sampler_size_est_need;
 252
 253
 254 /**
 255  * Percentage of total peer number in the gossip list
 256  * to send random PUSHes to
 257  */
 258 static float alpha;
 259
 260 /**
 261  * Percentage of total peer number in the gossip list
 262  * to send random PULLs to
 263  */
 264 static float beta;
 265
 266 /**
 267  * The percentage gamma of history updates.
 268  * Simply 1 - alpha - beta
 269  */
 270
 271
 272 /**
 273  * Identifier for the main task that runs periodically.
 274  */
 275 static struct GNUNET_SCHEDULER_Task *do_round_task;
 276
 277 /**
 278  * Time inverval the do_round task runs in.
 279  */
 280 static struct GNUNET_TIME_Relative round_interval;
 281
 282
 283
 284 /**
 285  * List to store peers received through pushes temporary.
 286  *
 287  * TODO -> multipeermap
 288  */
 289 static struct GNUNET_PeerIdentity *push_list;
 290
 291 /**
 292  * Size of the push_list;
 293  */
 294 static unsigned int push_list_size;
 295 //size_t push_list_size;
 296
 297 /**
 298  * List to store peers received through pulls temporary.
 299  *
 300  * TODO -> multipeermap
 301  */
 302 static struct GNUNET_PeerIdentity *pull_list;
 303
 304 /**
 305  * Size of the pull_list;
 306  */
 307 static unsigned int pull_list_size;
 308 //size_t pull_list_size;
 309
 310
 311 /**
 312  * Handler to NSE.
 313  */
 314 static struct GNUNET_NSE_Handle *nse;
 315
 316 /**
 317  * Handler to CADET.
 318  */
 319 static struct GNUNET_CADET_Handle *cadet_handle;
 320
 321 /**
 322  * Handler to PEERINFO.
 323  */
 324 static struct GNUNET_PEERINFO_Handle *peerinfo_handle;
 325
 326 /**
 327  * Handle for cancellation of iteration over peers.
 328  */
 329 struct GNUNET_PEERINFO_NotifyContext *peerinfo_notify_handle;
 330
 331
 332 /**
 333  * Request counter.
 334  *
 335  * Only needed in the beginning to check how many of the 64 deltas
 336  * we already have
 337  */
 338 static unsigned int req_counter;
 339
 340 /**
 341  * Time of the last request we received.
 342  *
 343  * Used to compute the expected request rate.
 344  */
 345 static struct GNUNET_TIME_Absolute last_request;
 346
 347 /**
 348  * Size of #request_deltas.
 349  */
 350 #define REQUEST_DELTAS_SIZE 64
 351 static unsigned int request_deltas_size = REQUEST_DELTAS_SIZE;
 352
 353 /**
 354  * Last 64 deltas between requests
 355  */
 356 static struct GNUNET_TIME_Relative request_deltas[REQUEST_DELTAS_SIZE];
 357
 358 /**
 359  * The prediction of the rate of requests
 360  */
 361 static struct GNUNET_TIME_Relative  request_rate;
 362
 363
 364 /**
 365  * List with the peers we sent requests to.
 366  */
 367 struct GNUNET_PeerIdentity *pending_pull_reply_list;
 368
 369 /**
 370  * Size of #pending_pull_reply_list.
 371  */
 372 uint32_t pending_pull_reply_list_size;
 373
 374
 375 /**
 376  * Number of history update tasks.
 377  */
 378 uint32_t num_hist_update_tasks;
 379
 380
 381 /**
 382  * Closure used to pass the client and the id to the callback
 383  * that replies to a client's request
 384  */
 385 struct ReplyCls
 386 {
 387   /**
 388    * The identifier of the request
 389    */
 390   uint32_t id;
 391
 392   /**
 393    * The client handle to send the reply to
 394    */
 395   struct GNUNET_SERVER_Client *client;
 396 };
 397
 398
 399 #ifdef ENABLE_MALICIOUS
 400 /**
 401  * Type of malicious peer
 402  *
 403  * 0 Don't act malicious at all - Default
 404  * 1 Try to maximise representation
 405  * 2 Try to partition the network
 406  * 3 Combined attack
 407  */
 408 uint32_t mal_type = 0;
 409
 410 /**
 411  * Other malicious peers
 412  */
 413 static struct GNUNET_PeerIdentity *mal_peers = NULL;
 414
 415 /**
 416  * Hashmap of malicious peers used as set.
 417  * Used to more efficiently check whether we know that peer.
 418  */
 419 static struct GNUNET_CONTAINER_MultiPeerMap *mal_peer_set = NULL;
 420
 421 /**
 422  * Number of other malicious peers
 423  */
 424 static uint32_t num_mal_peers = 0;
 425
 426
 427 /**
 428  * If type is 2 This struct is used to store the attacked peers in a DLL
 429  */
 430 struct AttackedPeer
 431 {
 432   /**
 433    * DLL
 434    */
 435   struct AttackedPeer *next;
 436   struct AttackedPeer *prev;
 437
 438   /**
 439    * PeerID
 440    */
 441   struct GNUNET_PeerIdentity peer_id;
 442 };
 443
 444 /**
 445  * If type is 2 this is the DLL of attacked peers
 446  */
 447 static struct AttackedPeer *att_peers_head = NULL;
 448 static struct AttackedPeer *att_peers_tail = NULL;
 449
 450 /**
 451  * This index is used to point to an attacked peer to
 452  * implement the round-robin-ish way to select attacked peers.
 453  */
 454 static struct AttackedPeer *att_peer_index = NULL;
 455
 456 /**
 457  * Hashmap of attacked peers used as set.
 458  * Used to more efficiently check whether we know that peer.
 459  */
 460 static struct GNUNET_CONTAINER_MultiPeerMap *att_peer_set = NULL;
 461
 462 /**
 463  * Number of attacked peers
 464  */
 465 static uint32_t num_attacked_peers = 0;
 466
 467
 468 /**
 469  * If type is 1 this is the attacked peer
 470  */
 471 static struct GNUNET_PeerIdentity attacked_peer;
 472
 473 /**
 474  * The limit of PUSHes we can send in one round.
 475  * This is an assumption of the Brahms protocol and either implemented
 476  * via proof of work
 477  * or
 478  * assumend to be the bandwidth limitation.
 479  */
 480 static uint32_t push_limit = 10000;
 481 #endif /* ENABLE_MALICIOUS */
 482
 483
 484 /***********************************************************************
 485  * /Globals
 486 ***********************************************************************/
 487
 488
 489
 490
 491
 492
 493 /***********************************************************************
 494  * Util functions
 495 ***********************************************************************/
 496
 497 /**
 498  * Set a peer flag of given peer context.
 499  */
 500 #define set_peer_flag(peer_ctx, mask) (peer_ctx->peer_flags |= mask)
 501
 502 /**
 503  * Get peer flag of given peer context.
 504  */
 505 #define get_peer_flag(peer_ctx, mask) (peer_ctx->peer_flags & mask ? GNUNET_YES : GNUNET_NO)
 506
 507 /**
 508  * Unset flag of given peer context.
 509  */
 510 #define unset_peer_flag(peer_ctx, mask) (peer_ctx->peer_flags &= (~mask))
 511
 512 /**
 513  * Clean the send channel of a peer
 514  */
 515 void
 516 peer_clean (const struct GNUNET_PeerIdentity *peer);
 517
 518
 519 /**
 520  * Check if peer is already in peer array.
 521  */
 522   int
 523 in_arr (const struct GNUNET_PeerIdentity *array,
 524         unsigned int arr_size,
 525         const struct GNUNET_PeerIdentity *peer)
 526 {
 527   GNUNET_assert (NULL != peer);
 528
 529   if (0 == arr_size)
 530     return GNUNET_NO;
 531
 532   GNUNET_assert (NULL != array);
 533
 534   unsigned int i;
 535
 536   for (i = 0; i < arr_size ; i++)
 537     if (0 == GNUNET_CRYPTO_cmp_peer_identity (&array[i], peer))
 538       return GNUNET_YES;
 539   return GNUNET_NO;
 540 }
 541
 542
 543 /**
 544  * Print peerlist to log.
 545  */
 546 void
 547 print_peer_list (struct GNUNET_PeerIdentity *list, unsigned int len)
 548 {
 549   unsigned int i;
 550
 551   LOG (GNUNET_ERROR_TYPE_DEBUG,
 552        "Printing peer list of length %u at %p:\n",
 553        len,
 554        list);
 555   for (i = 0 ; i < len ; i++)
 556   {
 557     LOG (GNUNET_ERROR_TYPE_DEBUG,
 558          "%u. peer: %s\n",
 559          i, GNUNET_i2s (&list[i]));
 560   }
 561 }
 562
 563
 564 /**
 565  * Remove peer from list.
 566  */
 567   void
 568 rem_from_list (struct GNUNET_PeerIdentity **peer_list,
 569                unsigned int *list_size,
 570                const struct GNUNET_PeerIdentity *peer)
 571 {
 572   unsigned int i;
 573   struct GNUNET_PeerIdentity *tmp;
 574
 575   tmp = *peer_list;
 576
 577   LOG (GNUNET_ERROR_TYPE_DEBUG,
 578        "Removing peer %s from list at %p\n",
 579        GNUNET_i2s (peer),
 580        tmp);
 581
 582   for ( i = 0 ; i < *list_size ; i++ )
 583   {
 584     if (0 == GNUNET_CRYPTO_cmp_peer_identity (&tmp[i], peer))
 585     {
 586       if (i < *list_size -1)
 587       { /* Not at the last entry -- shift peers left */
 588         memcpy (&tmp[i], &tmp[i +1],
 589                 ((*list_size) - i -1) * sizeof (struct GNUNET_PeerIdentity));
 590       }
 591       /* Remove last entry (should be now useless PeerID) */
 592       GNUNET_array_grow (tmp, *list_size, (*list_size) -1);
 593     }
 594   }
 595   *peer_list = tmp;
 596 }
 597
 598 /**
 599  * Get random peer from the given list but don't return one from the @a ignore_list.
 600  */
 601   struct GNUNET_PeerIdentity *
 602 get_rand_peer_ignore_list (const struct GNUNET_PeerIdentity *peer_list,
 603                            uint32_t list_size,
 604                            const struct GNUNET_PeerIdentity *ignore_list,
 605                            uint32_t ignore_size)
 606 {
 607   uint32_t r_index;
 608   uint32_t tmp_size;
 609   struct GNUNET_PeerIdentity *tmp_peer_list;
 610   struct GNUNET_PeerIdentity *peer;
 611
 612   GNUNET_assert (NULL != peer_list);
 613   if (0 == list_size)
 614     return NULL;
 615
 616   tmp_size = 0;
 617   tmp_peer_list = NULL;
 618   GNUNET_array_grow (tmp_peer_list, tmp_size, list_size);
 619   memcpy (tmp_peer_list,
 620           peer_list,
 621           list_size * sizeof (struct GNUNET_PeerIdentity));
 622   peer = GNUNET_new (struct GNUNET_PeerIdentity);
 623
 624   /**;
 625    * Choose the r_index of the peer we want to return
 626    * at random from the interval of the gossip list
 627    */
 628   r_index = GNUNET_CRYPTO_random_u64 (GNUNET_CRYPTO_QUALITY_STRONG,
 629                                       tmp_size);
 630   *peer = tmp_peer_list[r_index];
 631
 632   while (in_arr (ignore_list, ignore_size, peer))
 633   {
 634     rem_from_list (&tmp_peer_list, &tmp_size, peer);
 635
 636     print_peer_list (tmp_peer_list, tmp_size);
 637
 638     if (0 == tmp_size)
 639     {
 640       GNUNET_free (peer);
 641       return NULL;
 642     }
 643
 644     /**;
 645      * Choose the r_index of the peer we want to return
 646      * at random from the interval of the gossip list
 647      */
 648     r_index = GNUNET_CRYPTO_random_u64 (GNUNET_CRYPTO_QUALITY_STRONG,
 649                                         tmp_size);
 650     *peer = tmp_peer_list[r_index];
 651   }
 652
 653
 654   GNUNET_array_grow (tmp_peer_list, tmp_size, 0);
 655
 656   return peer;
 657 }
 658
 659
 660 /**
 661  * Get the context of a peer. If not existing, create.
 662  */
 663   struct PeerContext *
 664 get_peer_ctx (struct GNUNET_CONTAINER_MultiPeerMap *peer_map,
 665               const struct GNUNET_PeerIdentity *peer)
 666 {
 667   struct PeerContext *ctx;
 668
 669   if ( GNUNET_YES == GNUNET_CONTAINER_multipeermap_contains (peer_map, peer))
 670   {
 671     ctx = GNUNET_CONTAINER_multipeermap_get (peer_map, peer);
 672   }
 673   else
 674   {
 675     ctx = GNUNET_new (struct PeerContext);
 676     ctx->peer_flags = 0;
 677     ctx->mq = NULL;
 678     ctx->send_channel = NULL;
 679     ctx->recv_channel = NULL;
 680     ctx->outstanding_ops = NULL;
 681     ctx->num_outstanding_ops = 0;
 682     ctx->is_live_task = NULL;
 683     ctx->peer_id = *peer;
 684     (void) GNUNET_CONTAINER_multipeermap_put (peer_map, peer, ctx,
 685                                               GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_FAST);
 686   }
 687   return ctx;
 688 }
 689
 690
 691 /**
 692  * Put random peer from sampler into the gossip list as history update.
 693  */
 694   void
 695 hist_update (void *cls, struct GNUNET_PeerIdentity *ids, uint32_t num_peers)
 696 {
 697   GNUNET_assert (1 == num_peers);
 698
 699   if (gossip_list_size < sampler_size_est_need)
 700   {
 701     GNUNET_array_append (gossip_list, gossip_list_size, *ids);
 702     to_file (file_name_view_log,
 703              "+%s\t(hist)",
 704              GNUNET_i2s_full (ids));
 705   }
 706
 707   if (0 < num_hist_update_tasks)
 708     num_hist_update_tasks--;
 709 }
 710
 711
 712 /**
 713  * Set the peer flag to living and call the outstanding operations on this peer.
 714  */
 715 static size_t
 716 peer_is_live (struct PeerContext *peer_ctx)
 717 {
 718   struct GNUNET_PeerIdentity *peer;
 719
 720   /* Cancle is_live_task if still scheduled */
 721   if (NULL != peer_ctx->is_live_task)
 722   {
 723     GNUNET_CADET_notify_transmit_ready_cancel (peer_ctx->is_live_task);
 724     peer_ctx->is_live_task = NULL;
 725   }
 726
 727   peer = &peer_ctx->peer_id;
 728   set_peer_flag (peer_ctx, VALID);
 729
 730   LOG (GNUNET_ERROR_TYPE_DEBUG, "Peer %s is live\n", GNUNET_i2s (peer));
 731
 732   if (0 < peer_ctx->num_outstanding_ops)
 733   { /* Call outstanding operations */
 734     unsigned int i;
 735
 736     for (i = 0 ; i < peer_ctx->num_outstanding_ops ; i++)
 737       peer_ctx->outstanding_ops[i].op (peer_ctx->outstanding_ops[i].op_cls, peer);
 738     GNUNET_array_grow (peer_ctx->outstanding_ops, peer_ctx->num_outstanding_ops, 0);
 739   }
 740
 741   return 0;
 742 }
 743
 744
 745 /**
 746  * Callback that is called when a channel was effectively established.
 747  * This is given to ntfy_tmt_rdy and called when the channel was
 748  * successfully established.
 749  */
 750 static size_t
 751 cadet_ntfy_tmt_rdy_cb (void *cls, size_t size, void *buf)
 752 {
 753   struct PeerContext *peer_ctx = (struct PeerContext *) cls;
 754
 755   peer_ctx->is_live_task = NULL;
 756   LOG (GNUNET_ERROR_TYPE_DEBUG,
 757        "Set ->is_live_task = NULL for peer %s\n",
 758        GNUNET_i2s (&peer_ctx->peer_id));
 759
 760   if (NULL != buf
 761       && 0 != size)
 762   {
 763     peer_is_live (peer_ctx);
 764   }
 765   else
 766   {
 767     LOG (GNUNET_ERROR_TYPE_WARNING,
 768          "Problems establishing a connection to peer %s in order to check liveliness\n",
 769          GNUNET_i2s (&peer_ctx->peer_id));
 770     // TODO reschedule? cleanup?
 771   }
 772
 773   //if (NULL != peer_ctx->is_live_task)
 774   //{
 775   //  LOG (GNUNET_ERROR_TYPE_DEBUG,
 776   //       "Trying to cancle is_live_task for peer %s\n",
 777   //       GNUNET_i2s (&peer_ctx->peer_id));
 778   //  GNUNET_CADET_notify_transmit_ready_cancel (peer_ctx->is_live_task);
 779   //  peer_ctx->is_live_task = NULL;
 780   //}
 781
 782   return 0;
 783 }
 784
 785
 786 /**
 787  * Get the channel of a peer. If not existing, create.
 788  */
 789   struct GNUNET_CADET_Channel *
 790 get_channel (struct GNUNET_CONTAINER_MultiPeerMap *peer_map,
 791              const struct GNUNET_PeerIdentity *peer)
 792 {
 793   struct PeerContext *peer_ctx;
 794
 795   peer_ctx = get_peer_ctx (peer_map, peer);
 796
 797   if (NULL == peer_ctx->send_channel)
 798   {
 799     LOG (GNUNET_ERROR_TYPE_DEBUG,
 800          "Trying to establish channel to peer %s\n",
 801          GNUNET_i2s (peer));
 802
 803     peer_ctx->send_channel =
 804       GNUNET_CADET_channel_create (cadet_handle,
 805                                    NULL,
 806                                    peer,
 807                                    GNUNET_RPS_CADET_PORT,
 808                                    GNUNET_CADET_OPTION_RELIABLE);
 809
 810     // do I have to explicitly put it in the peer_map?
 811     (void) GNUNET_CONTAINER_multipeermap_put
 812       (peer_map,
 813        peer,
 814        peer_ctx,
 815        GNUNET_CONTAINER_MULTIHASHMAPOPTION_REPLACE);
 816   }
 817   return peer_ctx->send_channel;
 818 }
 819
 820
 821 /**
 822  * Get the message queue of a specific peer.
 823  *
 824  * If we already have a message queue open to this client,
 825  * simply return it, otherways create one.
 826  */
 827   struct GNUNET_MQ_Handle *
 828 get_mq (struct GNUNET_CONTAINER_MultiPeerMap *peer_map,
 829         const struct GNUNET_PeerIdentity *peer_id)
 830 {
 831   struct PeerContext *peer_ctx;
 832
 833   peer_ctx = get_peer_ctx (peer_map, peer_id);
 834
 835   GNUNET_assert (NULL == peer_ctx->is_live_task);
 836
 837   if (NULL == peer_ctx->mq)
 838   {
 839     (void) get_channel (peer_map, peer_id);
 840     peer_ctx->mq = GNUNET_CADET_mq_create (peer_ctx->send_channel);
 841     //do I have to explicitly put it in the peer_map?
 842     (void) GNUNET_CONTAINER_multipeermap_put (peer_map, peer_id, peer_ctx,
 843                                        GNUNET_CONTAINER_MULTIHASHMAPOPTION_REPLACE);
 844   }
 845   return peer_ctx->mq;
 846 }
 847
 848
 849 /**
 850  * Issue check whether peer is live
 851  *
 852  * @param peer_ctx the context of the peer
 853  */
 854 void
 855 check_peer_live (struct PeerContext *peer_ctx)
 856 {
 857   (void) get_channel (peer_map, &peer_ctx->peer_id);
 858   LOG (GNUNET_ERROR_TYPE_DEBUG,
 859        "Get informed about peer %s getting live\n",
 860        GNUNET_i2s (&peer_ctx->peer_id));
 861   if (NULL == peer_ctx->is_live_task)
 862   {
 863     peer_ctx->is_live_task =
 864         GNUNET_CADET_notify_transmit_ready (peer_ctx->send_channel,
 865                                             GNUNET_NO,
 866                                             GNUNET_TIME_UNIT_FOREVER_REL,
 867                                             sizeof (struct GNUNET_MessageHeader),
 868                                             cadet_ntfy_tmt_rdy_cb,
 869                                             peer_ctx);
 870   }
 871   else
 872   {
 873     LOG (GNUNET_ERROR_TYPE_DEBUG,
 874          "Already waiting for notification\n");
 875   }
 876 }
 877
 878
 879 /**
 880  * Sum all time relatives of an array.
 881   */
 882   struct GNUNET_TIME_Relative
 883 T_relative_sum (const struct GNUNET_TIME_Relative *rel_array, uint32_t arr_size)
 884 {
 885   struct GNUNET_TIME_Relative sum;
 886   uint32_t i;
 887
 888   sum = GNUNET_TIME_UNIT_ZERO;
 889   for ( i = 0 ; i < arr_size ; i++ )
 890   {
 891     sum = GNUNET_TIME_relative_add (sum, rel_array[i]);
 892   }
 893   return sum;
 894 }
 895
 896
 897 /**
 898  * Compute the average of given time relatives.
 899  */
 900   struct GNUNET_TIME_Relative
 901 T_relative_avg (const struct GNUNET_TIME_Relative *rel_array, uint32_t arr_size)
 902 {
 903   return GNUNET_TIME_relative_divide (T_relative_sum (rel_array, arr_size), arr_size);
 904 }
 905
 906
 907 /**
 908  * Insert PeerID in #pull_list
 909  *
 910  * Called once we know a peer is live.
 911  */
 912   void
 913 insert_in_pull_list (void *cls, const struct GNUNET_PeerIdentity *peer)
 914 {
 915   if (GNUNET_NO == in_arr (pull_list, pull_list_size, peer))
 916     GNUNET_array_append (pull_list, pull_list_size, *peer);
 917
 918   peer_clean (peer);
 919 }
 920
 921 /**
 922  * Check whether #insert_in_pull_list was already scheduled
 923  */
 924   int
 925 insert_in_pull_list_scheduled (const struct PeerContext *peer_ctx)
 926 {
 927   unsigned int i;
 928
 929   for ( i = 0 ; i < peer_ctx->num_outstanding_ops ; i++ )
 930     if (insert_in_pull_list == peer_ctx->outstanding_ops[i].op)
 931       return GNUNET_YES;
 932   return GNUNET_NO;
 933 }
 934
 935
 936 /**
 937  * Insert PeerID in #gossip_list
 938  *
 939  * Called once we know a peer is live.
 940  */
 941   void
 942 insert_in_gossip_list (void *cls, const struct GNUNET_PeerIdentity *peer)
 943 {
 944   if (GNUNET_NO == in_arr (gossip_list, gossip_list_size, peer))
 945   {
 946     GNUNET_array_append (gossip_list, gossip_list_size, *peer);
 947     to_file (file_name_view_log,
 948              "+%s\t(ins in gossip list)",
 949              GNUNET_i2s_full (peer));
 950   }
 951
 952   (void) get_channel (peer_map, peer);
 953 }
 954
 955 /**
 956  * Check whether #insert_in_gossip_list was already scheduled
 957  */
 958   int
 959 insert_in_gossip_list_scheduled (const struct PeerContext *peer_ctx)
 960 {
 961   unsigned int i;
 962
 963   for ( i = 0 ; i < peer_ctx->num_outstanding_ops ; i++ )
 964     if (insert_in_gossip_list == peer_ctx->outstanding_ops[i].op)
 965       return GNUNET_YES;
 966   return GNUNET_NO;
 967 }
 968
 969
 970 /**
 971  * Update sampler with given PeerID.
 972  */
 973   void
 974 insert_in_sampler (void *cls, const struct GNUNET_PeerIdentity *peer)
 975 {
 976   LOG (GNUNET_ERROR_TYPE_DEBUG,
 977        "Updating samplers with peer %s from insert_in_sampler()\n",
 978        GNUNET_i2s (peer));
 979   RPS_sampler_update (prot_sampler,   peer);
 980   RPS_sampler_update (client_sampler, peer);
 981 }
 982
 983
 984 /**
 985  * Check whether #insert_in_sampler was already scheduled
 986  */
 987 static int
 988 insert_in_sampler_scheduled (const struct PeerContext *peer_ctx)
 989 {
 990   unsigned int i;
 991
 992   for (i = 0 ; i < peer_ctx->num_outstanding_ops ; i++)
 993     if (insert_in_sampler== peer_ctx->outstanding_ops[i].op)
 994       return GNUNET_YES;
 995   return GNUNET_NO;
 996 }
 997
 998
 999 /**
1000  * Wrapper around #RPS_sampler_resize()
1001  *
1002  * If we do not have enough sampler elements, double current sampler size
1003  * If we have more than enough sampler elements, halv current sampler size
1004  */
1005 static void
1006 resize_wrapper (struct RPS_Sampler *sampler, uint32_t new_size)
1007 {
1008   unsigned int sampler_size;
1009
1010   // TODO statistics
1011   // TODO respect the min, max
1012   sampler_size = RPS_sampler_get_size (sampler);
1013   if (sampler_size > new_size * 4)
1014   { /* Shrinking */
1015     RPS_sampler_resize (sampler, sampler_size / 2);
1016   }
1017   else if (sampler_size < new_size)
1018   { /* Growing */
1019     RPS_sampler_resize (sampler, sampler_size * 2);
1020   }
1021   LOG (GNUNET_ERROR_TYPE_DEBUG, "sampler_size is now %u\n", sampler_size);
1022 }
1023
1024
1025 /**
1026  * Wrapper around #RPS_sampler_resize() resizing the client sampler
1027  */
1028 static void
1029 client_resize_wrapper ()
1030 {
1031   uint32_t bigger_size;
1032   unsigned int sampler_size;
1033
1034   // TODO statistics
1035
1036   sampler_size = RPS_sampler_get_size (client_sampler);
1037
1038   if (sampler_size_est_need > sampler_size_client_need)
1039     bigger_size = sampler_size_est_need;
1040   else
1041     bigger_size = sampler_size_client_need;
1042
1043   // TODO respect the min, max
1044   resize_wrapper (client_sampler, bigger_size);
1045   LOG (GNUNET_ERROR_TYPE_DEBUG, "sampler_size is now %u\n", sampler_size);
1046 }
1047
1048
1049 /**
1050  * Estimate request rate
1051  *
1052  * Called every time we receive a request from the client.
1053  */
1054   void
1055 est_request_rate()
1056 {
1057   struct GNUNET_TIME_Relative max_round_duration;
1058
1059   if (request_deltas_size > req_counter)
1060     req_counter++;
1061   if ( 1 < req_counter)
1062   {
1063     /* Shift last request deltas to the right */
1064     memcpy (&request_deltas[1],
1065         request_deltas,
1066         (req_counter - 1) * sizeof (struct GNUNET_TIME_Relative));
1067
1068     /* Add current delta to beginning */
1069     request_deltas[0] =
1070         GNUNET_TIME_absolute_get_difference (last_request,
1071                                              GNUNET_TIME_absolute_get ());
1072     request_rate = T_relative_avg (request_deltas, req_counter);
1073
1074     /* Compute the duration a round will maximally take */
1075     max_round_duration =
1076         GNUNET_TIME_relative_add (round_interval,
1077                                   GNUNET_TIME_relative_divide (round_interval, 2));
1078
1079     /* Set the estimated size the sampler has to have to
1080      * satisfy the current client request rate */
1081     sampler_size_client_need =
1082         max_round_duration.rel_value_us / request_rate.rel_value_us;
1083
1084     /* Resize the sampler */
1085     client_resize_wrapper ();
1086   }
1087   last_request = GNUNET_TIME_absolute_get ();
1088 }
1089
1090
1091 /**
1092  * Add all peers in @a peer_array to @a peer_map used as set.
1093  *
1094  * @param peer_array array containing the peers
1095  * @param num_peers number of peers in @peer_array
1096  * @param peer_map the peermap to use as set
1097  */
1098 static void
1099 add_peer_array_to_set (const struct GNUNET_PeerIdentity *peer_array,
1100                        unsigned int num_peers,
1101                        struct GNUNET_CONTAINER_MultiPeerMap *peer_map)
1102 {
1103   unsigned int i;
1104   if (NULL == peer_map)
1105   {
1106     LOG (GNUNET_ERROR_TYPE_WARNING,
1107          "Trying to add peers to an empty peermap.\n");
1108     return;
1109   }
1110
1111   for (i = 0 ; i < num_peers ; i++)
1112   {
1113     GNUNET_CONTAINER_multipeermap_put (peer_map,
1114                                        &peer_array[i],
1115                                        NULL,
1116                                        GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_FAST);
1117   }
1118 }
1119
1120
1121 /**
1122  * Send a PULL REPLY to @a peer_id
1123  *
1124  * @param peer_id the peer to send the reply to.
1125  * @param peer_ids the peers to send to @a peer_id
1126  * @param num_peer_ids the number of peers to send to @a peer_id
1127  */
1128 static void
1129 send_pull_reply (const struct GNUNET_PeerIdentity *peer_id,
1130                  const struct GNUNET_PeerIdentity *peer_ids,
1131                  unsigned int num_peer_ids)
1132 {
1133   uint32_t send_size;
1134   struct GNUNET_MQ_Handle *mq;
1135   struct GNUNET_MQ_Envelope *ev;
1136   struct GNUNET_RPS_P2P_PullReplyMessage *out_msg;
1137
1138   /* Compute actual size */
1139   send_size = sizeof (struct GNUNET_RPS_P2P_PullReplyMessage) +
1140               num_peer_ids * sizeof (struct GNUNET_PeerIdentity);
1141
1142   if (GNUNET_CONSTANTS_MAX_CADET_MESSAGE_SIZE < send_size)
1143     /* Compute number of peers to send
1144      * If too long, simply truncate */
1145     // TODO select random ones via permutation
1146     //      or even better: do good protocol design
1147     send_size =
1148       (GNUNET_CONSTANTS_MAX_CADET_MESSAGE_SIZE -
1149        sizeof (struct GNUNET_RPS_P2P_PullReplyMessage)) /
1150        sizeof (struct GNUNET_PeerIdentity);
1151   else
1152     send_size = num_peer_ids;
1153
1154   LOG (GNUNET_ERROR_TYPE_DEBUG,
1155       "PULL REQUEST from peer %s received, going to send %u peers\n",
1156       GNUNET_i2s (peer_id), send_size);
1157
1158   mq = get_mq (peer_map, peer_id);
1159
1160   ev = GNUNET_MQ_msg_extra (out_msg,
1161                             send_size * sizeof (struct GNUNET_PeerIdentity),
1162                             GNUNET_MESSAGE_TYPE_RPS_PP_PULL_REPLY);
1163   out_msg->num_peers = htonl (send_size);
1164   memcpy (&out_msg[1], peer_ids,
1165          send_size * sizeof (struct GNUNET_PeerIdentity));
1166
1167   GNUNET_MQ_send (mq, ev);
1168 }
1169
1170
1171 /**
1172  * This function is called on new peer_ids from 'external' sources
1173  * (client seed, cadet get_peers(), ...)
1174  *
1175  * @param peer_id the new peer_id
1176  */
1177 static void
1178 new_peer_id (const struct GNUNET_PeerIdentity *peer_id)
1179 {
1180   struct PeerOutstandingOp out_op;
1181   struct PeerContext *peer_ctx;
1182
1183   if (NULL != peer_id
1184       && 0 != GNUNET_CRYPTO_cmp_peer_identity (&own_identity, peer_id))
1185   {
1186     LOG (GNUNET_ERROR_TYPE_DEBUG,
1187         "Got peer_id %s (at %p, gossip_list_size: %u)\n",
1188         GNUNET_i2s (peer_id),
1189         peer_id,
1190         gossip_list_size);
1191
1192     peer_ctx = get_peer_ctx (peer_map, peer_id);
1193     if (GNUNET_YES != get_peer_flag (peer_ctx, VALID))
1194     {
1195       if (GNUNET_NO == insert_in_sampler_scheduled (peer_ctx))
1196       {
1197         out_op.op = insert_in_sampler;
1198         out_op.op_cls = NULL;
1199         GNUNET_array_append (peer_ctx->outstanding_ops,
1200                              peer_ctx->num_outstanding_ops,
1201                              out_op);
1202       }
1203
1204       if (GNUNET_NO == insert_in_gossip_list_scheduled (peer_ctx))
1205       {
1206         out_op.op = insert_in_gossip_list;
1207         out_op.op_cls = NULL;
1208         GNUNET_array_append (peer_ctx->outstanding_ops,
1209                              peer_ctx->num_outstanding_ops,
1210                              out_op);
1211       }
1212
1213       /* Trigger livelyness test on peer */
1214       check_peer_live (peer_ctx);
1215     }
1216     // else...?
1217
1218     // send push/pull to each of those peers?
1219   }
1220 }
1221
1222
1223 /***********************************************************************
1224  * /Util functions
1225 ***********************************************************************/
1226
1227
1228
1229
1230
1231 /**
1232  * Function called by NSE.
1233  *
1234  * Updates sizes of sampler list and gossip list and adapt those lists
1235  * accordingly.
1236  */
1237   void
1238 nse_callback (void *cls, struct GNUNET_TIME_Absolute timestamp,
1239               double logestimate, double std_dev)
1240 {
1241   double estimate;
1242   //double scale; // TODO this might go gloabal/config
1243
1244   LOG (GNUNET_ERROR_TYPE_DEBUG,
1245        "Received a ns estimate - logest: %f, std_dev: %f (old_size: %u)\n",
1246        logestimate, std_dev, RPS_sampler_get_size (prot_sampler));
1247   //scale = .01;
1248   estimate = GNUNET_NSE_log_estimate_to_n (logestimate);
1249   // GNUNET_NSE_log_estimate_to_n (logestimate);
1250   estimate = pow (estimate, 1.0 / 3);
1251   // TODO add if std_dev is a number
1252   // estimate += (std_dev * scale);
1253   if (2 < ceil (estimate))
1254   {
1255     LOG (GNUNET_ERROR_TYPE_DEBUG, "Changing estimate to %f\n", estimate);
1256     sampler_size_est_need = estimate;
1257   } else
1258     LOG (GNUNET_ERROR_TYPE_DEBUG, "Not using estimate %f\n", estimate);
1259
1260   /* If the NSE has changed adapt the lists accordingly */
1261   resize_wrapper (prot_sampler, sampler_size_est_need);
1262   client_resize_wrapper ();
1263 }
1264
1265
1266 /**
1267  * Callback called once the requested PeerIDs are ready.
1268  *
1269  * Sends those to the requesting client.
1270  */
1271 void client_respond (void *cls,
1272     struct GNUNET_PeerIdentity *peer_ids, uint32_t num_peers)
1273 {
1274   struct GNUNET_MQ_Envelope *ev;
1275   struct GNUNET_RPS_CS_ReplyMessage *out_msg;
1276   struct ReplyCls *reply_cls = (struct ReplyCls *) cls;
1277   uint32_t size_needed;
1278   struct ClientContext *cli_ctx;
1279
1280   LOG (GNUNET_ERROR_TYPE_DEBUG,
1281        "sampler returned %" PRIu32 " peers\n",
1282        num_peers);
1283
1284   size_needed = sizeof (struct GNUNET_RPS_CS_ReplyMessage) +
1285                 num_peers * sizeof (struct GNUNET_PeerIdentity);
1286
1287   GNUNET_assert (GNUNET_SERVER_MAX_MESSAGE_SIZE >= size_needed);
1288
1289   ev = GNUNET_MQ_msg_extra (out_msg,
1290                             num_peers * sizeof (struct GNUNET_PeerIdentity),
1291                             GNUNET_MESSAGE_TYPE_RPS_CS_REPLY);
1292   out_msg->num_peers = htonl (num_peers);
1293   out_msg->id = htonl (reply_cls->id);
1294
1295   memcpy (&out_msg[1],
1296           peer_ids,
1297           num_peers * sizeof (struct GNUNET_PeerIdentity));
1298   GNUNET_free (peer_ids);
1299
1300   cli_ctx = GNUNET_SERVER_client_get_user_context (reply_cls->client, struct ClientContext);
1301   if (NULL == cli_ctx) {
1302     cli_ctx = GNUNET_new (struct ClientContext);
1303     cli_ctx->mq = GNUNET_MQ_queue_for_server_client (reply_cls->client);
1304     GNUNET_SERVER_client_set_user_context (reply_cls->client, cli_ctx);
1305   }
1306
1307   GNUNET_free (reply_cls);
1308
1309   GNUNET_MQ_send (cli_ctx->mq, ev);
1310 }
1311
1312
1313 /**
1314  * Handle RPS request from the client.
1315  *
1316  * @param cls closure
1317  * @param client identification of the client
1318  * @param message the actual message
1319  */
1320 static void
1321 handle_client_request (void *cls,
1322                        struct GNUNET_SERVER_Client *client,
1323                        const struct GNUNET_MessageHeader *message)
1324 {
1325   struct GNUNET_RPS_CS_RequestMessage *msg;
1326   uint32_t num_peers;
1327   uint32_t size_needed;
1328   struct ReplyCls *reply_cls;
1329   uint32_t i;
1330
1331   msg = (struct GNUNET_RPS_CS_RequestMessage *) message;
1332
1333   num_peers = ntohl (msg->num_peers);
1334   size_needed = sizeof (struct GNUNET_RPS_CS_RequestMessage) +
1335                 num_peers * sizeof (struct GNUNET_PeerIdentity);
1336
1337   if (GNUNET_SERVER_MAX_MESSAGE_SIZE < size_needed)
1338   {
1339     GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
1340     return;
1341   }
1342
1343   for (i = 0 ; i < num_peers ; i++)
1344     est_request_rate();
1345
1346   LOG (GNUNET_ERROR_TYPE_DEBUG,
1347        "Client requested %" PRIu32 " random peer(s).\n",
1348        num_peers);
1349
1350   reply_cls = GNUNET_new (struct ReplyCls);
1351   reply_cls->id = ntohl (msg->id);
1352   reply_cls->client = client;
1353
1354   RPS_sampler_get_n_rand_peers (client_sampler,
1355                                 client_respond,
1356                                 reply_cls,
1357                                 num_peers);
1358
1359   GNUNET_SERVER_receive_done (client,
1360                               GNUNET_OK);
1361 }
1362
1363
1364 /**
1365  * Handle seed from the client.
1366  *
1367  * @param cls closure
1368  * @param client identification of the client
1369  * @param message the actual message
1370  */
1371   static void
1372 handle_client_seed (void *cls,
1373                     struct GNUNET_SERVER_Client *client,
1374                     const struct GNUNET_MessageHeader *message)
1375 {
1376   struct GNUNET_RPS_CS_SeedMessage *in_msg;
1377   struct GNUNET_PeerIdentity *peers;
1378   uint32_t num_peers;
1379   uint32_t i;
1380
1381   if (sizeof (struct GNUNET_RPS_CS_SeedMessage) > ntohs (message->size))
1382   {
1383     GNUNET_break_op (0);
1384     GNUNET_SERVER_receive_done (client,
1385                                 GNUNET_SYSERR);
1386   }
1387
1388   in_msg = (struct GNUNET_RPS_CS_SeedMessage *) message;
1389   num_peers = ntohl (in_msg->num_peers);
1390   peers = (struct GNUNET_PeerIdentity *) &in_msg[1];
1391   //peers = GNUNET_new_array (num_peers, struct GNUNET_PeerIdentity);
1392   //memcpy (peers, &in_msg[1], num_peers * sizeof (struct GNUNET_PeerIdentity));
1393
1394   if ((ntohs (message->size) - sizeof (struct GNUNET_RPS_CS_SeedMessage)) /
1395       sizeof (struct GNUNET_PeerIdentity) != num_peers)
1396   {
1397     GNUNET_break_op (0);
1398     GNUNET_SERVER_receive_done (client,
1399                                 GNUNET_SYSERR);
1400   }
1401
1402   LOG (GNUNET_ERROR_TYPE_DEBUG,
1403        "Client seeded peers:\n");
1404   print_peer_list (peers, num_peers);
1405
1406   for (i = 0 ; i < num_peers ; i++)
1407   {
1408     LOG (GNUNET_ERROR_TYPE_DEBUG,
1409          "Updating samplers with seed %" PRIu32 ": %s\n",
1410          i,
1411          GNUNET_i2s (&peers[i]));
1412
1413     new_peer_id (&peers[i]);
1414
1415     //RPS_sampler_update (prot_sampler,   &peers[i]);
1416     //RPS_sampler_update (client_sampler, &peers[i]);
1417   }
1418
1419   ////GNUNET_free (peers);
1420
1421   GNUNET_SERVER_receive_done (client,
1422                                                 GNUNET_OK);
1423 }
1424
1425
1426 /**
1427  * Handle a PUSH message from another peer.
1428  *
1429  * Check the proof of work and store the PeerID
1430  * in the temporary list for pushed PeerIDs.
1431  *
1432  * @param cls Closure
1433  * @param channel The channel the PUSH was received over
1434  * @param channel_ctx The context associated with this channel
1435  * @param msg The message header
1436  */
1437 static int
1438 handle_peer_push (void *cls,
1439     struct GNUNET_CADET_Channel *channel,
1440     void **channel_ctx,
1441     const struct GNUNET_MessageHeader *msg)
1442 {
1443   const struct GNUNET_PeerIdentity *peer;
1444
1445   // (check the proof of work)
1446
1447   peer = (const struct GNUNET_PeerIdentity *)
1448     GNUNET_CADET_channel_get_info (channel, GNUNET_CADET_OPTION_PEER);
1449   // FIXME wait for cadet to change this function
1450
1451   LOG (GNUNET_ERROR_TYPE_DEBUG, "PUSH received (%s)\n", GNUNET_i2s (peer));
1452
1453   #ifdef ENABLE_MALICIOUS
1454   struct AttackedPeer *tmp_att_peer;
1455
1456   tmp_att_peer = GNUNET_new (struct AttackedPeer);
1457   memcpy (&tmp_att_peer->peer_id, peer, sizeof (struct GNUNET_PeerIdentity));
1458   if (1 == mal_type
1459       || 3 == mal_type)
1460   { /* Try to maximise representation */
1461     if (NULL == att_peer_set)
1462       att_peer_set = GNUNET_CONTAINER_multipeermap_create (1, GNUNET_NO);
1463     if (GNUNET_NO == GNUNET_CONTAINER_multipeermap_contains (att_peer_set,
1464                                                              peer))
1465     {
1466       GNUNET_CONTAINER_DLL_insert (att_peers_head,
1467                                    att_peers_tail,
1468                                    tmp_att_peer);
1469       add_peer_array_to_set (peer, 1, att_peer_set);
1470     }
1471     return GNUNET_OK;
1472   }
1473
1474
1475   else if (2 == mal_type)
1476   { /* We attack one single well-known peer - simply ignore */
1477     return GNUNET_OK;
1478   }
1479   else
1480   {
1481     GNUNET_free (tmp_att_peer);
1482   }
1483
1484   #endif /* ENABLE_MALICIOUS */
1485
1486   /* Add the sending peer to the push_list */
1487   if (GNUNET_NO == in_arr (push_list, push_list_size, peer))
1488     GNUNET_array_append (push_list, push_list_size, *peer);
1489
1490   return GNUNET_OK;
1491 }
1492
1493
1494 /**
1495  * Handle PULL REQUEST request message from another peer.
1496  *
1497  * Reply with the gossip list of PeerIDs.
1498  *
1499  * @param cls Closure
1500  * @param channel The channel the PUSH was received over
1501  * @param channel_ctx The context associated with this channel
1502  * @param msg The message header
1503  */
1504 static int
1505 handle_peer_pull_request (void *cls,
1506     struct GNUNET_CADET_Channel *channel,
1507     void **channel_ctx,
1508     const struct GNUNET_MessageHeader *msg)
1509 {
1510   struct GNUNET_PeerIdentity *peer;
1511
1512   peer = (struct GNUNET_PeerIdentity *)
1513     GNUNET_CADET_channel_get_info (channel,
1514                                    GNUNET_CADET_OPTION_PEER);
1515   // FIXME wait for cadet to change this function
1516
1517   #ifdef ENABLE_MALICIOUS
1518   if (1 == mal_type
1519       || 3 == mal_type)
1520   { /* Try to maximise representation */
1521     send_pull_reply (peer, mal_peers, num_mal_peers);
1522     return GNUNET_OK;
1523   }
1524
1525   else if (2 == mal_type)
1526   { /* Try to partition network */
1527     if (GNUNET_YES == GNUNET_CRYPTO_cmp_peer_identity (&attacked_peer, peer))
1528     {
1529       send_pull_reply (peer, mal_peers, num_mal_peers);
1530     }
1531     return GNUNET_OK;
1532   }
1533   #endif /* ENABLE_MALICIOUS */
1534
1535   send_pull_reply (peer, gossip_list, gossip_list_size);
1536
1537   return GNUNET_OK;
1538 }
1539
1540
1541 /**
1542  * Handle PULL REPLY message from another peer.
1543  *
1544  * Check whether we sent a corresponding request and
1545  * whether this reply is the first one.
1546  *
1547  * @param cls Closure
1548  * @param channel The channel the PUSH was received over
1549  * @param channel_ctx The context associated with this channel
1550  * @param msg The message header
1551  */
1552   static int
1553 handle_peer_pull_reply (void *cls,
1554                         struct GNUNET_CADET_Channel *channel,
1555                         void **channel_ctx,
1556                         const struct GNUNET_MessageHeader *msg)
1557 {
1558   struct GNUNET_RPS_P2P_PullReplyMessage *in_msg;
1559   struct GNUNET_PeerIdentity *peers;
1560   struct PeerContext *peer_ctx;
1561   struct GNUNET_PeerIdentity *sender;
1562   struct PeerContext *sender_ctx;
1563   struct PeerOutstandingOp out_op;
1564   uint32_t i;
1565 #ifdef ENABLE_MALICIOUS
1566   struct AttackedPeer *tmp_att_peer;
1567 #endif /* ENABLE_MALICIOUS */
1568
1569   /* Check for protocol violation */
1570   if (sizeof (struct GNUNET_RPS_P2P_PullReplyMessage) > ntohs (msg->size))
1571   {
1572     GNUNET_break_op (0);
1573     return GNUNET_SYSERR;
1574   }
1575
1576   in_msg = (struct GNUNET_RPS_P2P_PullReplyMessage *) msg;
1577   if ((ntohs (msg->size) - sizeof (struct GNUNET_RPS_P2P_PullReplyMessage)) /
1578       sizeof (struct GNUNET_PeerIdentity) != ntohl (in_msg->num_peers))
1579   {
1580     LOG (GNUNET_ERROR_TYPE_ERROR,
1581         "message says it sends %" PRIu64 " peers, have space for %i peers\n",
1582         ntohl (in_msg->num_peers),
1583         (ntohs (msg->size) - sizeof (struct GNUNET_RPS_P2P_PullReplyMessage)) /
1584             sizeof (struct GNUNET_PeerIdentity));
1585     GNUNET_break_op (0);
1586     return GNUNET_SYSERR;
1587   }
1588
1589   sender = (struct GNUNET_PeerIdentity *) GNUNET_CADET_channel_get_info (
1590       (struct GNUNET_CADET_Channel *) channel, GNUNET_CADET_OPTION_PEER);
1591        // Guess simply casting isn't the nicest way...
1592        // FIXME wait for cadet to change this function
1593   sender_ctx = get_peer_ctx (peer_map, sender);
1594
1595   if (GNUNET_YES == get_peer_flag (sender_ctx, PULL_REPLY_PENDING))
1596   {
1597     GNUNET_break_op (0);
1598     return GNUNET_OK;
1599   }
1600
1601
1602   #ifdef ENABLE_MALICIOUS
1603   // We shouldn't even receive pull replies as we're not sending
1604   if (2 == mal_type)
1605     return GNUNET_OK;
1606   #endif /* ENABLE_MALICIOUS */
1607
1608   /* Do actual logic */
1609   peers = (struct GNUNET_PeerIdentity *) &msg[1];
1610
1611   LOG (GNUNET_ERROR_TYPE_DEBUG,
1612        "PULL REPLY received, got following peers:\n");
1613
1614   for (i = 0 ; i < ntohl (in_msg->num_peers) ; i++)
1615   {
1616     LOG (GNUNET_ERROR_TYPE_DEBUG,
1617          "%u. %s\n",
1618          i,
1619          GNUNET_i2s (&peers[i]));
1620
1621     #ifdef ENABLE_MALICIOUS
1622     if (1 == mal_type
1623         || 3 == mal_type)
1624     { /* Add attacked peer to local list */
1625       // TODO check if we sent a request and this was the first reply
1626       if (GNUNET_NO == GNUNET_CONTAINER_multipeermap_contains (att_peer_set,
1627                                                                &peers[i])
1628           && GNUNET_NO == GNUNET_CONTAINER_multipeermap_contains (mal_peer_set,
1629                                                                   &peers[i])
1630           && GNUNET_NO == GNUNET_CRYPTO_cmp_peer_identity (&peers[i],
1631                                                            &own_identity))
1632       {
1633         tmp_att_peer = GNUNET_new (struct AttackedPeer);
1634         tmp_att_peer->peer_id = peers[i];
1635         GNUNET_CONTAINER_DLL_insert (att_peers_head,
1636                                      att_peers_tail,
1637                                      tmp_att_peer);
1638         add_peer_array_to_set (&peers[i], 1, att_peer_set);
1639       }
1640       continue;
1641     }
1642     #endif /* ENABLE_MALICIOUS */
1643     peer_ctx = get_peer_ctx (peer_map, &peers[i]);
1644     if (GNUNET_YES == get_peer_flag (peer_ctx, VALID)
1645         || NULL != peer_ctx->send_channel
1646         || NULL != peer_ctx->recv_channel)
1647     {
1648       if (GNUNET_NO == in_arr (pull_list, pull_list_size, &peers[i])
1649           && 0 != GNUNET_CRYPTO_cmp_peer_identity (&own_identity, &peers[i]))
1650         GNUNET_array_append (pull_list, pull_list_size, peers[i]);
1651     }
1652     else if (GNUNET_NO == insert_in_pull_list_scheduled (peer_ctx))
1653     {
1654       out_op.op = insert_in_pull_list;
1655       out_op.op_cls = NULL;
1656       GNUNET_array_append (peer_ctx->outstanding_ops,
1657                            peer_ctx->num_outstanding_ops,
1658                            out_op);
1659       check_peer_live (peer_ctx);
1660     }
1661   }
1662
1663   unset_peer_flag (sender_ctx, PULL_REPLY_PENDING);
1664   rem_from_list (&pending_pull_reply_list, &pending_pull_reply_list_size, sender);
1665
1666   return GNUNET_OK;
1667 }
1668
1669
1670 /**
1671  * Compute a random delay.
1672  * A uniformly distributed value between mean + spread and mean - spread.
1673  *
1674  * For example for mean 4 min and spread 2 the minimum is (4 min - (1/2 * 4 min))
1675  * It would return a random value between 2 and 6 min.
1676  *
1677  * @param mean the mean
1678  * @param spread the inverse amount of deviation from the mean
1679  */
1680 static struct GNUNET_TIME_Relative
1681 compute_rand_delay (struct GNUNET_TIME_Relative mean, unsigned int spread)
1682 {
1683   struct GNUNET_TIME_Relative half_interval;
1684   struct GNUNET_TIME_Relative ret;
1685   unsigned int rand_delay;
1686   unsigned int max_rand_delay;
1687
1688   if (0 == spread)
1689   {
1690     LOG (GNUNET_ERROR_TYPE_WARNING,
1691          "Not accepting spread of 0\n");
1692     GNUNET_break (0);
1693   }
1694
1695   /* Compute random time value between spread * mean and spread * mean */
1696   half_interval = GNUNET_TIME_relative_divide (mean, spread);
1697
1698   max_rand_delay = GNUNET_TIME_UNIT_FOREVER_REL.rel_value_us / mean.rel_value_us * (2/spread);
1699   /**
1700    * Compute random value between (0 and 1) * round_interval
1701    * via multiplying round_interval with a 'fraction' (0 to value)/value
1702    */
1703   rand_delay = GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK, max_rand_delay);
1704   ret = GNUNET_TIME_relative_multiply (mean,  rand_delay);
1705   ret = GNUNET_TIME_relative_divide   (ret, max_rand_delay);
1706   ret = GNUNET_TIME_relative_add      (ret, half_interval);
1707
1708   if (GNUNET_TIME_UNIT_FOREVER_REL.rel_value_us == ret.rel_value_us)
1709     LOG (GNUNET_ERROR_TYPE_WARNING,
1710          "Returning FOREVER_REL\n");
1711
1712   return ret;
1713 }
1714
1715
1716 /**
1717  * Send single pull request
1718  *
1719  * @param peer_id the peer to send the pull request to.
1720  */
1721 static void
1722 send_pull_request (struct GNUNET_PeerIdentity *peer_id)
1723 {
1724   struct GNUNET_MQ_Envelope *ev;
1725   struct GNUNET_MQ_Handle *mq;
1726
1727   LOG (GNUNET_ERROR_TYPE_DEBUG,
1728        "Sending PULL request to peer %s of gossiped list.\n",
1729        GNUNET_i2s (peer_id));
1730
1731   GNUNET_array_append (pending_pull_reply_list, pending_pull_reply_list_size, *peer_id);
1732
1733   ev = GNUNET_MQ_msg_header (GNUNET_MESSAGE_TYPE_RPS_PP_PULL_REQUEST);
1734   mq = get_mq (peer_map, peer_id);
1735   GNUNET_MQ_send (mq, ev);
1736 }
1737
1738
1739 /**
1740  * Send single push
1741  *
1742  * @param peer_id the peer to send the push to.
1743  */
1744 static void
1745 send_push (struct GNUNET_PeerIdentity *peer_id)
1746 {
1747   struct GNUNET_MQ_Envelope *ev;
1748   struct GNUNET_MQ_Handle *mq;
1749
1750   LOG (GNUNET_ERROR_TYPE_DEBUG,
1751        "Sending PUSH to peer %s of gossiped list.\n",
1752        GNUNET_i2s (peer_id));
1753
1754   ev = GNUNET_MQ_msg_header (GNUNET_MESSAGE_TYPE_RPS_PP_PUSH);
1755   mq = get_mq (peer_map, peer_id);
1756   GNUNET_MQ_send (mq, ev);
1757 }
1758
1759
1760 static void
1761 do_round (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc);
1762
1763 static void
1764 do_mal_round (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc);
1765
1766
1767 #ifdef ENABLE_MALICIOUS
1768 /**
1769  * Turn RPS service to act malicious.
1770  *
1771  * @param cls Closure
1772  * @param channel The channel the PUSH was received over
1773  * @param channel_ctx The context associated with this channel
1774  * @param msg The message header
1775  */
1776   static void
1777 handle_client_act_malicious (void *cls,
1778                              struct GNUNET_SERVER_Client *client,
1779                              const struct GNUNET_MessageHeader *msg)
1780 {
1781   struct GNUNET_RPS_CS_ActMaliciousMessage *in_msg;
1782   struct GNUNET_PeerIdentity *peers;
1783   uint32_t num_mal_peers_sent;
1784   uint32_t num_mal_peers_old;
1785
1786   /* Check for protocol violation */
1787   if (sizeof (struct GNUNET_RPS_CS_ActMaliciousMessage) > ntohs (msg->size))
1788   {
1789     GNUNET_break_op (0);
1790   }
1791
1792   in_msg = (struct GNUNET_RPS_CS_ActMaliciousMessage *) msg;
1793   if ((ntohs (msg->size) - sizeof (struct GNUNET_RPS_CS_ActMaliciousMessage)) /
1794       sizeof (struct GNUNET_PeerIdentity) != ntohl (in_msg->num_peers))
1795   {
1796     LOG (GNUNET_ERROR_TYPE_ERROR,
1797         "message says it sends %" PRIu64 " peers, have space for %i peers\n",
1798         ntohl (in_msg->num_peers),
1799         (ntohs (msg->size) - sizeof (struct GNUNET_RPS_CS_ActMaliciousMessage)) /
1800             sizeof (struct GNUNET_PeerIdentity));
1801     GNUNET_break_op (0);
1802   }
1803
1804
1805   /* Do actual logic */
1806   peers = (struct GNUNET_PeerIdentity *) &msg[1];
1807   mal_type = ntohl (in_msg->type);
1808
1809   LOG (GNUNET_ERROR_TYPE_DEBUG,
1810        "Now acting malicious type %" PRIu32 "\n",
1811        mal_type);
1812
1813   if (1 == mal_type)
1814   { /* Try to maximise representation */
1815     /* Add other malicious peers to those we already know */
1816
1817     num_mal_peers_sent = ntohl (in_msg->num_peers);
1818     num_mal_peers_old = num_mal_peers;
1819     GNUNET_array_grow (mal_peers,
1820                        num_mal_peers,
1821                        num_mal_peers + num_mal_peers_sent);
1822     memcpy (&mal_peers[num_mal_peers_old],
1823             peers,
1824             num_mal_peers_sent * sizeof (struct GNUNET_PeerIdentity));
1825
1826     /* Add all mal peers to mal_peer_set */
1827     add_peer_array_to_set (&mal_peers[num_mal_peers_old],
1828                            num_mal_peers_sent,
1829                            mal_peer_set);
1830
1831     /* Substitute do_round () with do_mal_round () */
1832     GNUNET_SCHEDULER_cancel (do_round_task);
1833     do_round_task = GNUNET_SCHEDULER_add_now (&do_mal_round, NULL);
1834   }
1835
1836   else if (2 == mal_type
1837            || 3 == mal_type)
1838   { /* Try to partition the network */
1839     /* Add other malicious peers to those we already know */
1840     num_mal_peers_sent = ntohl (in_msg->num_peers) - 1;
1841     num_mal_peers_old = num_mal_peers;
1842     GNUNET_array_grow (mal_peers,
1843                        num_mal_peers,
1844                        num_mal_peers + num_mal_peers_sent);
1845     memcpy (&mal_peers[num_mal_peers_old],
1846             peers,
1847             num_mal_peers_sent * sizeof (struct GNUNET_PeerIdentity));
1848
1849     /* Add all mal peers to mal_peer_set */
1850     add_peer_array_to_set (&mal_peers[num_mal_peers_old],
1851                            num_mal_peers_sent,
1852                            mal_peer_set);
1853
1854     /* Store the one attacked peer */
1855     memcpy (&attacked_peer,
1856             &in_msg->attacked_peer,
1857             sizeof (struct GNUNET_PeerIdentity));
1858
1859     LOG (GNUNET_ERROR_TYPE_DEBUG,
1860          "Attacked peer is %s\n",
1861          GNUNET_i2s (&attacked_peer));
1862
1863     /* Substitute do_round () with do_mal_round () */
1864     GNUNET_SCHEDULER_cancel (do_round_task);
1865     do_round_task = GNUNET_SCHEDULER_add_now (&do_mal_round, NULL);
1866   }
1867   else if (0 == mal_type)
1868   { /* Stop acting malicious */
1869     GNUNET_array_grow (mal_peers, num_mal_peers, 0);
1870
1871     /* Substitute do_mal_round () with do_round () */
1872     GNUNET_SCHEDULER_cancel (do_round_task);
1873     do_round_task = GNUNET_SCHEDULER_add_now (&do_round, NULL);
1874   }
1875   else
1876   {
1877     GNUNET_break (0);
1878   }
1879
1880   GNUNET_SERVER_receive_done (client,   GNUNET_OK);
1881 }
1882
1883
1884 /**
1885  * Send out PUSHes and PULLs maliciously.
1886  *
1887  * This is executed regylary.
1888  */
1889 static void
1890 do_mal_round (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
1891 {
1892   uint32_t num_pushes;
1893   uint32_t i;
1894   struct GNUNET_TIME_Relative time_next_round;
1895   struct AttackedPeer *tmp_att_peer;
1896
1897   LOG (GNUNET_ERROR_TYPE_DEBUG, "Going to execute next round maliciously.\n");
1898
1899   /* Do malicious actions */
1900   if (1 == mal_type)
1901   { /* Try to maximise representation */
1902
1903     /* The maximum of pushes we're going to send this round */
1904     num_pushes = GNUNET_MIN (GNUNET_MIN (push_limit,
1905                                          num_attacked_peers),
1906                              GNUNET_CONSTANTS_MAX_CADET_MESSAGE_SIZE);
1907
1908     LOG (GNUNET_ERROR_TYPE_DEBUG,
1909          "Going to send %" PRIu32 " pushes\n",
1910          num_pushes);
1911
1912     /* Send PUSHes to attacked peers */
1913     for (i = 0 ; i < num_pushes ; i++)
1914     {
1915       if (att_peers_tail == att_peer_index)
1916         att_peer_index = att_peers_head;
1917       else
1918         att_peer_index = att_peer_index->next;
1919
1920       send_push (&att_peer_index->peer_id);
1921     }
1922
1923     /* Send PULLs to some peers to learn about additional peers to attack */
1924     tmp_att_peer = att_peer_index;
1925     for (i = 0 ; i < num_pushes * alpha ; i++)
1926     {
1927       if (att_peers_tail == tmp_att_peer)
1928         tmp_att_peer = att_peers_head;
1929       else
1930         att_peer_index = tmp_att_peer->next;
1931
1932       send_pull_request (&tmp_att_peer->peer_id);
1933     }
1934   }
1935
1936
1937   else if (2 == mal_type)
1938   { /**
1939      * Try to partition the network
1940      * Send as many pushes to the attacked peer as possible
1941      * That is one push per round as it will ignore more.
1942      */
1943       send_push (&attacked_peer);
1944   }
1945
1946
1947   if (3 == mal_type)
1948   { /* Combined attack */
1949
1950     /* The maximum of pushes we're going to send this round */
1951     num_pushes = GNUNET_MIN (GNUNET_MIN (push_limit - 1,
1952                                          num_attacked_peers),
1953                              GNUNET_CONSTANTS_MAX_CADET_MESSAGE_SIZE);
1954
1955     LOG (GNUNET_ERROR_TYPE_DEBUG,
1956          "Going to send %" PRIu32 " pushes\n",
1957          num_pushes);
1958
1959     /* Send PUSHes to attacked peers */
1960     send_push (&attacked_peer);
1961
1962     for (i = 0 ; i < num_pushes ; i++)
1963     {
1964       if (att_peers_tail == att_peer_index)
1965         att_peer_index = att_peers_head;
1966       else
1967         att_peer_index = att_peer_index->next;
1968
1969       send_push (&att_peer_index->peer_id);
1970     }
1971
1972     /* Send PULLs to some peers to learn about additional peers to attack */
1973     tmp_att_peer = att_peer_index;
1974     for (i = 0 ; i < num_pushes * alpha ; i++)
1975     {
1976       if (att_peers_tail == tmp_att_peer)
1977         tmp_att_peer = att_peers_head;
1978       else
1979         att_peer_index = tmp_att_peer->next;
1980
1981       send_pull_request (&tmp_att_peer->peer_id);
1982     }
1983   }
1984
1985   /* Schedule next round */
1986   time_next_round = compute_rand_delay (round_interval, 2);
1987
1988   //do_round_task = GNUNET_SCHEDULER_add_delayed (round_interval, &do_mal_round, NULL);
1989   do_round_task = GNUNET_SCHEDULER_add_delayed (time_next_round, &do_mal_round, NULL);
1990   LOG (GNUNET_ERROR_TYPE_DEBUG, "Finished round\n");
1991 }
1992 #endif /* ENABLE_MALICIOUS */
1993
1994
1995 /**
1996  * Send out PUSHes and PULLs, possibly update #gossip_list, samplers.
1997  *
1998  * This is executed regylary.
1999  */
2000 static void
2001 do_round (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
2002 {
2003   LOG (GNUNET_ERROR_TYPE_DEBUG, "Going to execute next round.\n");
2004
2005   uint32_t i;
2006   unsigned int *permut;
2007   unsigned int n_peers; /* Number of peers we send pushes/pulls to */
2008   struct GNUNET_PeerIdentity peer;
2009   struct GNUNET_PeerIdentity *tmp_peer;
2010
2011   LOG (GNUNET_ERROR_TYPE_DEBUG,
2012        "Printing gossip list:\n");
2013   to_file (file_name_view_log,
2014            "___ new round ___");
2015   for (i = 0 ; i < gossip_list_size ; i++)
2016   {
2017     LOG (GNUNET_ERROR_TYPE_DEBUG,
2018          "\t%s\n", GNUNET_i2s (&gossip_list[i]));
2019     to_file (file_name_view_log,
2020              "=%s\t(do round)",
2021              GNUNET_i2s_full (&gossip_list[i]));
2022   }
2023   // TODO log lists, ...
2024
2025   /* Would it make sense to have one shuffeled gossip list and then
2026    * to send PUSHes to first alpha peers, PULL requests to next beta peers and
2027    * use the rest to update sampler?
2028    * in essence get random peers with consumption */
2029
2030   /* Send PUSHes */
2031   if (0 < gossip_list_size)
2032   {
2033     permut = GNUNET_CRYPTO_random_permute (GNUNET_CRYPTO_QUALITY_STRONG,
2034                                            (unsigned int) gossip_list_size);
2035     n_peers = ceil (alpha * gossip_list_size);
2036     LOG (GNUNET_ERROR_TYPE_DEBUG,
2037          "Going to send pushes to %u (ceil (%f * %u)) peers.\n",
2038          n_peers, alpha, gossip_list_size);
2039     for (i = 0 ; i < n_peers ; i++)
2040     {
2041       peer = gossip_list[permut[i]];
2042       if (0 != GNUNET_CRYPTO_cmp_peer_identity (&own_identity, &peer)) // TODO
2043       { // FIXME if this fails schedule/loop this for later
2044         send_push (&peer);
2045       }
2046     }
2047     GNUNET_free (permut);
2048   }
2049
2050
2051   /* Send PULL requests */
2052   //permut = GNUNET_CRYPTO_random_permute (GNUNET_CRYPTO_QUALITY_STRONG, (unsigned int) sampler_list->size);
2053   n_peers = ceil (beta * gossip_list_size);
2054   LOG (GNUNET_ERROR_TYPE_DEBUG,
2055        "Going to send pulls to %u (ceil (%f * %u)) peers.\n",
2056        n_peers, beta, gossip_list_size);
2057   for (i = 0 ; i < n_peers ; i++)
2058   {
2059     tmp_peer = get_rand_peer_ignore_list (gossip_list, gossip_list_size,
2060         pending_pull_reply_list, pending_pull_reply_list_size);
2061     if (NULL != tmp_peer)
2062     {
2063       peer = *tmp_peer;
2064       GNUNET_free (tmp_peer);
2065
2066       if (0 != GNUNET_CRYPTO_cmp_peer_identity (&own_identity, &peer))
2067       {
2068         send_pull_request (&peer);
2069       }
2070     }
2071   }
2072
2073
2074   /* Update gossip list */
2075   /* TODO see how many peers are in push-/pull- list! */
2076
2077   if (push_list_size <= alpha * gossip_list_size
2078       && push_list_size > 0
2079       && pull_list_size > 0)
2080   {
2081     LOG (GNUNET_ERROR_TYPE_DEBUG, "Update of the gossip list.\n");
2082
2083     uint32_t first_border;
2084     uint32_t second_border;
2085     uint32_t r_index;
2086     uint32_t peers_to_clean_size;
2087     struct GNUNET_PeerIdentity *peers_to_clean;
2088
2089     peers_to_clean = NULL;
2090     peers_to_clean_size = 0;
2091     GNUNET_array_grow (peers_to_clean, peers_to_clean_size, gossip_list_size);
2092     memcpy (peers_to_clean,
2093             gossip_list,
2094             gossip_list_size * sizeof (struct GNUNET_PeerIdentity));
2095
2096     first_border  =                ceil (alpha * sampler_size_est_need);
2097     second_border = first_border + ceil (beta  * sampler_size_est_need);
2098
2099     GNUNET_array_grow (gossip_list, gossip_list_size, second_border);
2100
2101     to_file (file_name_view_log,
2102              "--- emptied ---");
2103
2104     for (i = 0 ; i < first_border ; i++)
2105     {/* Update gossip list with peers received through PUSHes */
2106       r_index = GNUNET_CRYPTO_random_u64 (GNUNET_CRYPTO_QUALITY_STRONG,
2107                                           push_list_size);
2108       gossip_list[i] = push_list[r_index];
2109       to_file (file_name_view_log,
2110                "+%s't(push list)",
2111                GNUNET_i2s_full (&gossip_list[i]));
2112       // TODO change the peer_flags accordingly
2113     }
2114
2115     for (i = first_border ; i < second_border ; i++)
2116     {/* Update gossip list with peers received through PULLs */
2117       r_index = GNUNET_CRYPTO_random_u64 (GNUNET_CRYPTO_QUALITY_STRONG,
2118                                           pull_list_size);
2119       gossip_list[i] = pull_list[r_index];
2120       to_file (file_name_view_log,
2121                "+%s\t(pull list)",
2122                GNUNET_i2s_full (&gossip_list[i]));
2123       // TODO change the peer_flags accordingly
2124     }
2125
2126     for (i = second_border ; i < sampler_size_est_need ; i++)
2127     {/* Update gossip list with peers from history */
2128       RPS_sampler_get_n_rand_peers (prot_sampler,
2129                                     hist_update,
2130                                     NULL,
2131                                     1);
2132       num_hist_update_tasks++;
2133       // TODO change the peer_flags accordingly
2134     }
2135
2136     for (i = 0 ; i < gossip_list_size ; i++)
2137       rem_from_list (&peers_to_clean, &peers_to_clean_size, &gossip_list[i]);
2138
2139     for (i = 0 ; i < peers_to_clean_size ; i++)
2140     {
2141       peer_clean (&peers_to_clean[i]);
2142       /* to_file (file_name_view_log,
2143                "-%s",
2144                GNUNET_i2s_full (&peers_to_clean[i])); */
2145     }
2146
2147     GNUNET_free (peers_to_clean);
2148   }
2149   else
2150   {
2151     LOG (GNUNET_ERROR_TYPE_DEBUG, "No update of the gossip list.\n");
2152   }
2153   // TODO independent of that also get some peers from CADET_get_peers()?
2154
2155   LOG (GNUNET_ERROR_TYPE_DEBUG,
2156        "Received %u pushes and %u pulls last round (alpha (%.2f) * gossip_list_size (%u) = %.2f)\n",
2157        push_list_size,
2158        pull_list_size,
2159        alpha,
2160        gossip_list_size,
2161        alpha * gossip_list_size);
2162
2163   /* Update samplers */
2164   for ( i = 0 ; i < push_list_size ; i++ )
2165   {
2166     LOG (GNUNET_ERROR_TYPE_DEBUG,
2167          "Updating with peer %s from push list\n",
2168          GNUNET_i2s (&push_list[i]));
2169     RPS_sampler_update (prot_sampler,   &push_list[i]);
2170     RPS_sampler_update (client_sampler, &push_list[i]);
2171     // TODO set in_flag?
2172   }
2173
2174   for ( i = 0 ; i < pull_list_size ; i++ )
2175   {
2176     LOG (GNUNET_ERROR_TYPE_DEBUG,
2177          "Updating with peer %s from pull list\n",
2178          GNUNET_i2s (&pull_list[i]));
2179     RPS_sampler_update (prot_sampler,   &pull_list[i]);
2180     RPS_sampler_update (client_sampler, &pull_list[i]);
2181     // TODO set in_flag?
2182   }
2183
2184
2185   /* Empty push/pull lists */
2186   GNUNET_array_grow (push_list, push_list_size, 0);
2187   GNUNET_array_grow (pull_list, pull_list_size, 0);
2188
2189   struct GNUNET_TIME_Relative time_next_round;
2190
2191   time_next_round = compute_rand_delay (round_interval, 2);
2192
2193   /* Schedule next round */
2194   //do_round_task = GNUNET_SCHEDULER_add_delayed (round_interval, &do_round, NULL);
2195   do_round_task = GNUNET_SCHEDULER_add_delayed (time_next_round, &do_round, NULL);
2196   LOG (GNUNET_ERROR_TYPE_DEBUG, "Finished round\n");
2197 }
2198
2199
2200 static void
2201 rps_start (struct GNUNET_SERVER_Handle *server);
2202
2203
2204 /**
2205  * This is called from GNUNET_CADET_get_peers().
2206  *
2207  * It is called on every peer(ID) that cadet somehow has contact with.
2208  * We use those to initialise the sampler.
2209  */
2210 void
2211 init_peer_cb (void *cls,
2212               const struct GNUNET_PeerIdentity *peer,
2213               int tunnel, // "Do we have a tunnel towards this peer?"
2214               unsigned int n_paths, // "Number of known paths towards this peer"
2215               unsigned int best_path) // "How long is the best path?
2216                                       // (0 = unknown, 1 = ourselves, 2 = neighbor)"
2217 {
2218   if (NULL != peer)
2219   {
2220     LOG (GNUNET_ERROR_TYPE_DEBUG,
2221          "Got peer_id %s from cadet\n",
2222          GNUNET_i2s (peer));
2223     new_peer_id (peer);
2224   }
2225 }
2226
2227
2228 /**
2229  * Iterator over peers from peerinfo.
2230  *
2231  * @param cls closure
2232  * @param peer id of the peer, NULL for last call
2233  * @param hello hello message for the peer (can be NULL)
2234  * @param error message
2235  */
2236 void
2237 process_peerinfo_peers (void *cls,
2238                         const struct GNUNET_PeerIdentity *peer,
2239                         const struct GNUNET_HELLO_Message *hello,
2240                         const char *err_msg)
2241 {
2242   if (NULL != peer)
2243   {
2244     LOG (GNUNET_ERROR_TYPE_DEBUG,
2245          "Got peer_id %s from peerinfo\n",
2246          GNUNET_i2s (peer));
2247     new_peer_id (peer);
2248   }
2249 }
2250
2251
2252 /**
2253  * Clean the send channel of a peer
2254  */
2255 void
2256 peer_clean (const struct GNUNET_PeerIdentity *peer)
2257 {
2258   struct PeerContext *peer_ctx;
2259   struct GNUNET_CADET_Channel *channel;
2260
2261   if (GNUNET_YES != in_arr (gossip_list, gossip_list_size, peer)
2262       && GNUNET_YES == GNUNET_CONTAINER_multipeermap_contains (peer_map, peer))
2263   {
2264     peer_ctx = get_peer_ctx (peer_map, peer);
2265     if (NULL != peer_ctx->send_channel)
2266     {
2267       channel = peer_ctx->send_channel;
2268       peer_ctx->send_channel = NULL;
2269       GNUNET_CADET_channel_destroy (channel);
2270     }
2271   }
2272 }
2273
2274
2275 /**
2276  * Callback used to remove peers from the multipeermap.
2277  */
2278   int
2279 peer_remove_cb (void *cls, const struct GNUNET_PeerIdentity *key, void *value)
2280 {
2281   struct PeerContext *peer_ctx;
2282   const struct GNUNET_CADET_Channel *channel =
2283     (const struct GNUNET_CADET_Channel *) cls;
2284   struct GNUNET_CADET_Channel *recv;
2285   struct GNUNET_CADET_Channel *send;
2286
2287   peer_ctx = (struct PeerContext *) value;
2288
2289   LOG (GNUNET_ERROR_TYPE_DEBUG,
2290        "Going to clean peer %s\n",
2291        GNUNET_i2s (&peer_ctx->peer_id));
2292
2293   /* If operations are still scheduled for this peer cancel those */
2294   if (0 != peer_ctx->num_outstanding_ops)
2295     GNUNET_array_grow (peer_ctx->outstanding_ops,
2296                        peer_ctx->num_outstanding_ops,
2297                        0);
2298
2299   /* If we are still waiting for notification whether this peer is live
2300    * cancel the according task */
2301   if (NULL != peer_ctx->is_live_task)
2302   {
2303   LOG (GNUNET_ERROR_TYPE_DEBUG,
2304        "Trying to cancle is_live_task for peer %s\n",
2305        GNUNET_i2s (key));
2306     GNUNET_CADET_notify_transmit_ready_cancel (peer_ctx->is_live_task);
2307     peer_ctx->is_live_task = NULL;
2308   }
2309
2310
2311   recv = peer_ctx->recv_channel;
2312   peer_ctx->recv_channel = NULL;
2313   if (NULL != recv
2314       && channel != recv)
2315   {
2316     GNUNET_CADET_channel_destroy (recv);
2317   }
2318
2319   /* If there is still a mq destroy it */
2320   if (NULL != peer_ctx->mq)
2321   {
2322     GNUNET_MQ_destroy (peer_ctx->mq);
2323     peer_ctx->mq = NULL;
2324   }
2325
2326   /* Remove the send_channel
2327    * The peer map entry should be removed
2328    * from the callback #cleanup_channel */
2329   send = peer_ctx->send_channel;
2330   peer_ctx->send_channel = NULL;
2331   if (NULL != send
2332       && channel != send)
2333   {
2334     set_peer_flag (peer_ctx, TO_DESTROY);
2335     GNUNET_CADET_channel_destroy (send);
2336   }
2337   else
2338   { /* If there is no channel we have to remove it now */
2339     if (GNUNET_YES != GNUNET_CONTAINER_multipeermap_remove_all (peer_map, key))
2340       LOG (GNUNET_ERROR_TYPE_WARNING, "removing peer from peer_map failed\n");
2341     else
2342       GNUNET_free (peer_ctx);
2343   }
2344
2345   return GNUNET_YES;
2346 }
2347
2348
2349 /**
2350  * Task run during shutdown.
2351  *
2352  * @param cls unused
2353  * @param tc unused
2354  */
2355 static void
2356 shutdown_task (void *cls,
2357                      const struct GNUNET_SCHEDULER_TaskContext *tc)
2358 {
2359
2360   LOG (GNUNET_ERROR_TYPE_DEBUG, "RPS is going down\n");
2361
2362   GNUNET_PEERINFO_notify_cancel (peerinfo_notify_handle);
2363   GNUNET_PEERINFO_disconnect (peerinfo_handle);
2364
2365   if (NULL != do_round_task)
2366   {
2367     GNUNET_SCHEDULER_cancel (do_round_task);
2368     do_round_task = NULL;
2369   }
2370
2371   {
2372   if (GNUNET_SYSERR ==
2373         GNUNET_CONTAINER_multipeermap_iterate (peer_map, peer_remove_cb, NULL))
2374     LOG (GNUNET_ERROR_TYPE_WARNING,
2375         "Iterating over peers to disconnect from them was cancelled\n");
2376   }
2377
2378   GNUNET_NSE_disconnect (nse);
2379   RPS_sampler_destroy (prot_sampler);
2380   RPS_sampler_destroy (client_sampler);
2381   LOG (GNUNET_ERROR_TYPE_DEBUG,
2382        "Size of the peermap: %u\n",
2383        GNUNET_CONTAINER_multipeermap_size (peer_map));
2384   GNUNET_break (0 == GNUNET_CONTAINER_multipeermap_size (peer_map));
2385   GNUNET_CONTAINER_multipeermap_destroy (peer_map);
2386   GNUNET_CADET_disconnect (cadet_handle);
2387   GNUNET_array_grow (gossip_list, gossip_list_size, 0);
2388   GNUNET_array_grow (push_list, push_list_size, 0);
2389   GNUNET_array_grow (pull_list, pull_list_size, 0);
2390   #ifdef ENABLE_MALICIOUS
2391   struct AttackedPeer *tmp_att_peer;
2392   GNUNET_array_grow (mal_peers, num_mal_peers, 0);
2393   if (NULL != mal_peer_set)
2394     GNUNET_CONTAINER_multipeermap_destroy (mal_peer_set);
2395   if (NULL != att_peer_set)
2396     GNUNET_CONTAINER_multipeermap_destroy (att_peer_set);
2397   while (NULL != att_peers_head)
2398   {
2399     tmp_att_peer = att_peers_head;
2400     GNUNET_CONTAINER_DLL_remove (att_peers_head, att_peers_tail, tmp_att_peer);
2401   }
2402   #endif /* ENABLE_MALICIOUS */
2403 }
2404
2405
2406 /**
2407  * A client disconnected.  Remove all of its data structure entries.
2408  *
2409  * @param cls closure, NULL
2410  * @param client identification of the client
2411  */
2412 static void
2413 handle_client_disconnect (void *cls,
2414                           struct GNUNET_SERVER_Client * client)
2415 {
2416 }
2417
2418
2419 /**
2420  * Handle the channel a peer opens to us.
2421  *
2422  * @param cls The closure
2423  * @param channel The channel the peer wants to establish
2424  * @param initiator The peer's peer ID
2425  * @param port The port the channel is being established over
2426  * @param options Further options
2427  */
2428   static void *
2429 handle_inbound_channel (void *cls,
2430                         struct GNUNET_CADET_Channel *channel,
2431                         const struct GNUNET_PeerIdentity *initiator,
2432                         uint32_t port,
2433                         enum GNUNET_CADET_ChannelOption options)
2434 {
2435   struct PeerContext *peer_ctx;
2436   struct GNUNET_PeerIdentity peer;
2437
2438   peer = *initiator;
2439   LOG (GNUNET_ERROR_TYPE_DEBUG,
2440       "New channel was established to us (Peer %s).\n",
2441       GNUNET_i2s (&peer));
2442
2443   GNUNET_assert (NULL != channel);
2444
2445   // we might not even store the recv_channel
2446
2447   peer_ctx = get_peer_ctx (peer_map, &peer);
2448   // FIXME what do we do if a channel is established twice?
2449   //       overwrite? Clean old channel? ...?
2450   //if (NULL != peer_ctx->recv_channel)
2451   //{
2452   //  peer_ctx->recv_channel = channel;
2453   //}
2454   peer_ctx->recv_channel = channel;
2455
2456   (void) GNUNET_CONTAINER_multipeermap_put (peer_map, &peer, peer_ctx,
2457       GNUNET_CONTAINER_MULTIHASHMAPOPTION_REPLACE);
2458
2459   peer_is_live (peer_ctx);
2460
2461   return NULL; // TODO
2462 }
2463
2464
2465 /**
2466  * This is called when a remote peer destroys a channel.
2467  *
2468  * @param cls The closure
2469  * @param channel The channel being closed
2470  * @param channel_ctx The context associated with this channel
2471  */
2472   static void
2473 cleanup_channel (void *cls,
2474                 const struct GNUNET_CADET_Channel *channel,
2475                 void *channel_ctx)
2476 {
2477   struct GNUNET_PeerIdentity *peer;
2478   struct PeerContext *peer_ctx;
2479
2480   peer = (struct GNUNET_PeerIdentity *) GNUNET_CADET_channel_get_info (
2481       (struct GNUNET_CADET_Channel *) channel, GNUNET_CADET_OPTION_PEER);
2482        // Guess simply casting isn't the nicest way...
2483        // FIXME wait for cadet to change this function
2484
2485   //if (GNUNET_YES == GNUNET_CONTAINER_multipeermap_contains (peer_map, peer))
2486   //{
2487     peer_ctx = GNUNET_CONTAINER_multipeermap_get (peer_map, peer);
2488
2489     if (NULL == peer_ctx) /* It could have been removed by shutdown_task */
2490       return;
2491
2492
2493     if (channel == peer_ctx->send_channel)
2494     { /* Something (but us) killd the channel */
2495       LOG (GNUNET_ERROR_TYPE_DEBUG,
2496            "send channel (%s) was destroyed - cleaning up\n",
2497            GNUNET_i2s (peer));
2498
2499       rem_from_list (&gossip_list,
2500                      &gossip_list_size,
2501                      peer);
2502       rem_from_list (&pending_pull_reply_list,
2503                      &pending_pull_reply_list_size,
2504                      peer);
2505       to_file (file_name_view_log,
2506                "-%s\t(cleanup channel, other peer)",
2507                GNUNET_i2s_full (peer));
2508
2509       peer_ctx->send_channel = NULL;
2510       /* Somwewhat {ab,re}use the iterator function */
2511       /* Cast to void is ok, because it's used as void in peer_remove_cb */
2512       (void) peer_remove_cb ((void *) channel, peer, peer_ctx);
2513
2514       //if (GNUNET_YES != GNUNET_CONTAINER_multipeermap_remove_all (peer_map, key))
2515       //  LOG (GNUNET_ERROR_TYPE_WARNING, "Removing peer from peer_map failed\n");
2516       //else
2517       //  GNUNET_free (peer_ctx);
2518     }
2519     else if (channel == peer_ctx->recv_channel)
2520     { /* Other peer doesn't want to send us messages anymore */
2521       LOG (GNUNET_ERROR_TYPE_DEBUG,
2522            "Peer %s destroyed recv channel - cleaning up channel\n",
2523            GNUNET_i2s (peer));
2524       peer_ctx->recv_channel = NULL;
2525     }
2526     else if (NULL == peer_ctx->send_channel &&
2527              get_peer_flag (peer_ctx, TO_DESTROY))
2528     { /* We closed the channel to that peer */
2529       LOG (GNUNET_ERROR_TYPE_DEBUG,
2530            "send channel (%s) was destroyed by us - cleaning up\n",
2531            GNUNET_i2s (peer));
2532
2533       rem_from_list (&gossip_list,
2534                      &gossip_list_size,
2535                      peer);
2536       rem_from_list (&pending_pull_reply_list,
2537                      &pending_pull_reply_list_size,
2538                      peer);
2539       to_file (file_name_view_log,
2540                "-%s\t(cleanup channel, ourself)",
2541                GNUNET_i2s_full (peer));
2542
2543       /* Somwewhat {ab,re}use the iterator function */
2544       /* Cast to void is ok, because it's used as void in peer_remove_cb */
2545       (void) peer_remove_cb ((void *) channel, peer, peer_ctx);
2546     }
2547     else
2548     {
2549       LOG (GNUNET_ERROR_TYPE_WARNING,
2550            "unknown channel (%s) was destroyed\n",
2551            GNUNET_i2s (peer));
2552     }
2553   //}
2554 }
2555
2556
2557 /**
2558  * Actually start the service.
2559  */
2560   static void
2561 rps_start (struct GNUNET_SERVER_Handle *server)
2562 {
2563   static const struct GNUNET_SERVER_MessageHandler handlers[] = {
2564     {&handle_client_request,     NULL, GNUNET_MESSAGE_TYPE_RPS_CS_REQUEST,
2565       sizeof (struct GNUNET_RPS_CS_RequestMessage)},
2566     {&handle_client_seed,        NULL, GNUNET_MESSAGE_TYPE_RPS_CS_SEED, 0},
2567     #ifdef ENABLE_MALICIOUS
2568     {&handle_client_act_malicious, NULL, GNUNET_MESSAGE_TYPE_RPS_ACT_MALICIOUS , 0},
2569     #endif /* ENABLE_MALICIOUS */
2570     {NULL, NULL, 0, 0}
2571   };
2572
2573   GNUNET_SERVER_add_handlers (server, handlers);
2574   GNUNET_SERVER_disconnect_notify (server,
2575                                    &handle_client_disconnect,
2576                                    NULL);
2577   LOG (GNUNET_ERROR_TYPE_INFO, "Ready to receive requests from clients\n");
2578
2579
2580   do_round_task = GNUNET_SCHEDULER_add_now (&do_round, NULL);
2581   LOG (GNUNET_ERROR_TYPE_DEBUG, "Scheduled first round\n");
2582
2583   GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_FOREVER_REL,
2584                                                         &shutdown_task,
2585                                                         NULL);
2586 }
2587
2588
2589 /**
2590  * Process statistics requests.
2591  *
2592  * @param cls closure
2593  * @param server the initialized server
2594  * @param c configuration to use
2595  */
2596   static void
2597 run (void *cls,
2598      struct GNUNET_SERVER_Handle *server,
2599      const struct GNUNET_CONFIGURATION_Handle *c)
2600 {
2601   int size;
2602   int out_size;
2603
2604   // TODO check what this does -- copied from gnunet-boss
2605   // - seems to work as expected
2606   GNUNET_log_setup ("rps", GNUNET_error_type_to_string (GNUNET_ERROR_TYPE_DEBUG), NULL);
2607   cfg = c;
2608
2609
2610   /* Get own ID */
2611   GNUNET_CRYPTO_get_peer_identity (cfg, &own_identity); // TODO check return value
2612   GNUNET_log (GNUNET_ERROR_TYPE_INFO,
2613               "STARTING SERVICE (rps) for peer [%s]\n",
2614               GNUNET_i2s (&own_identity));
2615   #ifdef ENABLE_MALICIOUS
2616   GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
2617               "Malicious execution compiled in.\n");
2618   #endif /* ENABLE_MALICIOUS */
2619
2620
2621
2622   /* Get time interval from the configuration */
2623   if (GNUNET_OK != GNUNET_CONFIGURATION_get_value_time (cfg, "RPS",
2624                                                         "ROUNDINTERVAL",
2625                                                         &round_interval))
2626   {
2627     LOG (GNUNET_ERROR_TYPE_DEBUG, "Failed to read ROUNDINTERVAL from config\n");
2628     GNUNET_SCHEDULER_shutdown ();
2629     return;
2630   }
2631
2632   /* Get initial size of sampler/gossip list from the configuration */
2633   if (GNUNET_OK != GNUNET_CONFIGURATION_get_value_number (cfg, "RPS",
2634                                                          "INITSIZE",
2635                                                          (long long unsigned int *) &sampler_size_est_need))
2636   {
2637     LOG (GNUNET_ERROR_TYPE_DEBUG, "Failed to read INITSIZE from config\n");
2638     GNUNET_SCHEDULER_shutdown ();
2639     return;
2640   }
2641   LOG (GNUNET_ERROR_TYPE_DEBUG, "INITSIZE is %" PRIu64 "\n", sampler_size_est_need);
2642
2643
2644   gossip_list = NULL;
2645   gossip_list_size = 0;
2646
2647   /* file_name_view_log */
2648   if (GNUNET_OK != GNUNET_DISK_directory_create ("/tmp/rps/"))
2649   {
2650     LOG (GNUNET_ERROR_TYPE_WARNING,
2651          "Failed to create directory /tmp/rps/\n");
2652   }
2653
2654   size = (14 + strlen (GNUNET_i2s_full (&own_identity)) + 1) * sizeof (char);
2655   file_name_view_log = GNUNET_malloc (size);
2656   out_size = GNUNET_snprintf (file_name_view_log,
2657                               size,
2658                               "/tmp/rps/view-%s",
2659                               GNUNET_i2s_full (&own_identity));
2660   if (size < out_size ||
2661       0 > out_size)
2662   {
2663     LOG (GNUNET_ERROR_TYPE_WARNING,
2664          "Failed to write string to buffer (size: %i, out_size: %i)\n",
2665          size,
2666          out_size);
2667   }
2668
2669
2670   /* connect to NSE */
2671   nse = GNUNET_NSE_connect (cfg, nse_callback, NULL);
2672
2673
2674   alpha = 0.45;
2675   beta  = 0.45;
2676
2677   peer_map = GNUNET_CONTAINER_multipeermap_create (sampler_size_est_need, GNUNET_NO);
2678
2679
2680   /* Initialise cadet */
2681   static const struct GNUNET_CADET_MessageHandler cadet_handlers[] = {
2682     {&handle_peer_push        , GNUNET_MESSAGE_TYPE_RPS_PP_PUSH        ,
2683       sizeof (struct GNUNET_MessageHeader)},
2684     {&handle_peer_pull_request, GNUNET_MESSAGE_TYPE_RPS_PP_PULL_REQUEST,
2685       sizeof (struct GNUNET_MessageHeader)},
2686     {&handle_peer_pull_reply  , GNUNET_MESSAGE_TYPE_RPS_PP_PULL_REPLY  , 0},
2687     {NULL, 0, 0}
2688   };
2689
2690   const uint32_t ports[] = {GNUNET_RPS_CADET_PORT, 0}; // _PORT specified in src/rps/rps.h
2691   cadet_handle = GNUNET_CADET_connect (cfg,
2692                                        cls,
2693                                        &handle_inbound_channel,
2694                                        &cleanup_channel,
2695                                        cadet_handlers,
2696                                        ports);
2697
2698   peerinfo_handle = GNUNET_PEERINFO_connect (cfg);
2699
2700   /* Initialise sampler */
2701   struct GNUNET_TIME_Relative half_round_interval;
2702   struct GNUNET_TIME_Relative  max_round_interval;
2703
2704   half_round_interval = GNUNET_TIME_relative_multiply (round_interval, .5);
2705   max_round_interval = GNUNET_TIME_relative_add (round_interval, half_round_interval);
2706
2707   prot_sampler =   RPS_sampler_init     (sampler_size_est_need, max_round_interval);
2708   client_sampler = RPS_sampler_mod_init (sampler_size_est_need, max_round_interval);
2709
2710   /* Initialise push and pull maps */
2711   push_list = NULL;
2712   push_list_size = 0;
2713   pull_list = NULL;
2714   pull_list_size = 0;
2715   pending_pull_reply_list = NULL;
2716   pending_pull_reply_list_size = 0;
2717
2718
2719   num_hist_update_tasks = 0;
2720
2721
2722   LOG (GNUNET_ERROR_TYPE_DEBUG, "Requesting peers from CADET\n");
2723   GNUNET_CADET_get_peers (cadet_handle, &init_peer_cb, NULL);
2724   // TODO send push/pull to each of those peers?
2725
2726   peerinfo_notify_handle = GNUNET_PEERINFO_notify (cfg,
2727                                                    GNUNET_NO,
2728                                                    process_peerinfo_peers,
2729                                                    NULL);
2730
2731   rps_start (server);
2732 }
2733
2734
2735 /**
2736  * The main function for the rps service.
2737  *
2738  * @param argc number of arguments from the command line
2739  * @param argv command line arguments
2740  * @return 0 ok, 1 on error
2741  */
2742   int
2743 main (int argc, char *const *argv)
2744 {
2745   return (GNUNET_OK ==
2746           GNUNET_SERVICE_run (argc,
2747                               argv,
2748                               "rps",
2749                               GNUNET_SERVICE_OPTION_NONE,
2750                               &run, NULL)) ? 0 : 1;
2751 }
2752
2753 /* end of gnunet-service-rps.c */