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