src/rps/gnunet-service-rps.c

   1 /*
   2      This file is part of GNUnet.
   3      Copyright (C) 2013-2015 GNUnet e.V.
   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., 51 Franklin Street, Fifth Floor,
  18      Boston, MA 02110-1301, 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_applications.h"
  28 #include "gnunet_util_lib.h"
  29 #include "gnunet_cadet_service.h"
  30 #include "gnunet_peerinfo_service.h"
  31 #include "gnunet_nse_service.h"
  32 #include "rps.h"
  33 #include "rps-test_util.h"
  34 #include "gnunet-service-rps_sampler.h"
  35 #include "gnunet-service-rps_custommap.h"
  36 #include "gnunet-service-rps_peers.h"
  37 #include "gnunet-service-rps_view.h"
  38
  39 #include <math.h>
  40 #include <inttypes.h>
  41
  42 #define LOG(kind, ...) GNUNET_log(kind, __VA_ARGS__)
  43
  44 // TODO modify @brief in every file
  45
  46 // TODO check for overflows
  47
  48 // TODO align message structs
  49
  50 // TODO connect to friends
  51
  52 // TODO store peers somewhere persistent
  53
  54 // TODO blacklist? (-> mal peer detection on top of brahms)
  55
  56 // hist_size_init, hist_size_max
  57
  58 /**
  59  * Our configuration.
  60  */
  61 static const struct GNUNET_CONFIGURATION_Handle *cfg;
  62
  63 /**
  64  * Our own identity.
  65  */
  66 static struct GNUNET_PeerIdentity own_identity;
  67
  68
  69 /***********************************************************************
  70  * Housekeeping with clients
  71 ***********************************************************************/
  72
  73 /**
  74  * Closure used to pass the client and the id to the callback
  75  * that replies to a client's request
  76  */
  77 struct ReplyCls
  78 {
  79   /**
  80    * DLL
  81    */
  82   struct ReplyCls *next;
  83   struct ReplyCls *prev;
  84
  85   /**
  86    * The identifier of the request
  87    */
  88   uint32_t id;
  89
  90   /**
  91    * The handle to the request
  92    */
  93   struct RPS_SamplerRequestHandle *req_handle;
  94
  95   /**
  96    * The client handle to send the reply to
  97    */
  98   struct ClientContext *cli_ctx;
  99 };
 100
 101
 102 /**
 103  * Struct used to store the context of a connected client.
 104  */
 105 struct ClientContext
 106 {
 107   /**
 108    * DLL
 109    */
 110   struct ClientContext *next;
 111   struct ClientContext *prev;
 112
 113   /**
 114    * The message queue to communicate with the client.
 115    */
 116   struct GNUNET_MQ_Handle *mq;
 117
 118   /**
 119    * DLL with handles to single requests from the client
 120    */
 121   struct ReplyCls *rep_cls_head;
 122   struct ReplyCls *rep_cls_tail;
 123
 124   /**
 125    * The client handle to send the reply to
 126    */
 127   struct GNUNET_SERVICE_Client *client;
 128 };
 129
 130 /**
 131  * DLL with all clients currently connected to us
 132  */
 133 struct ClientContext *cli_ctx_head;
 134 struct ClientContext *cli_ctx_tail;
 135
 136 /***********************************************************************
 137  * /Housekeeping with clients
 138 ***********************************************************************/
 139
 140
 141
 142
 143
 144 /***********************************************************************
 145  * Globals
 146 ***********************************************************************/
 147
 148 /**
 149  * Sampler used for the Brahms protocol itself.
 150  */
 151 static struct RPS_Sampler *prot_sampler;
 152
 153 /**
 154  * Sampler used for the clients.
 155  */
 156 static struct RPS_Sampler *client_sampler;
 157
 158 /**
 159  * Name to log view to
 160  */
 161 static char *file_name_view_log;
 162
 163 /**
 164  * The size of sampler we need to be able to satisfy the client's need
 165  * of random peers.
 166  */
 167 static unsigned int sampler_size_client_need;
 168
 169 /**
 170  * The size of sampler we need to be able to satisfy the Brahms protocol's
 171  * need of random peers.
 172  *
 173  * This is one minimum size the sampler grows to.
 174  */
 175 static unsigned int sampler_size_est_need;
 176
 177 /**
 178  * Percentage of total peer number in the view
 179  * to send random PUSHes to
 180  */
 181 static float alpha;
 182
 183 /**
 184  * Percentage of total peer number in the view
 185  * to send random PULLs to
 186  */
 187 static float beta;
 188
 189 /**
 190  * Identifier for the main task that runs periodically.
 191  */
 192 static struct GNUNET_SCHEDULER_Task *do_round_task;
 193
 194 /**
 195  * Time inverval the do_round task runs in.
 196  */
 197 static struct GNUNET_TIME_Relative round_interval;
 198
 199 /**
 200  * List to store peers received through pushes temporary.
 201  */
 202 static struct CustomPeerMap *push_map;
 203
 204 /**
 205  * List to store peers received through pulls temporary.
 206  */
 207 static struct CustomPeerMap *pull_map;
 208
 209 /**
 210  * Handler to NSE.
 211  */
 212 static struct GNUNET_NSE_Handle *nse;
 213
 214 /**
 215  * Handler to CADET.
 216  */
 217 static struct GNUNET_CADET_Handle *cadet_handle;
 218
 219 /**
 220  * Handler to PEERINFO.
 221  */
 222 static struct GNUNET_PEERINFO_Handle *peerinfo_handle;
 223
 224 /**
 225  * Handle for cancellation of iteration over peers.
 226  */
 227 static struct GNUNET_PEERINFO_NotifyContext *peerinfo_notify_handle;
 228
 229 /**
 230  * Request counter.
 231  *
 232  * Counts how many requets clients already issued.
 233  * Only needed in the beginning to check how many of the 64 deltas
 234  * we already have
 235  */
 236 static unsigned int req_counter;
 237
 238 /**
 239  * Time of the last request we received.
 240  *
 241  * Used to compute the expected request rate.
 242  */
 243 static struct GNUNET_TIME_Absolute last_request;
 244
 245 /**
 246  * Size of #request_deltas.
 247  */
 248 #define REQUEST_DELTAS_SIZE 64
 249 static unsigned int request_deltas_size = REQUEST_DELTAS_SIZE;
 250
 251 /**
 252  * Last 64 deltas between requests
 253  */
 254 static struct GNUNET_TIME_Relative request_deltas[REQUEST_DELTAS_SIZE];
 255
 256 /**
 257  * The prediction of the rate of requests
 258  */
 259 static struct GNUNET_TIME_Relative request_rate;
 260
 261
 262 #ifdef ENABLE_MALICIOUS
 263 /**
 264  * Type of malicious peer
 265  *
 266  * 0 Don't act malicious at all - Default
 267  * 1 Try to maximise representation
 268  * 2 Try to partition the network
 269  * 3 Combined attack
 270  */
 271 static uint32_t mal_type;
 272
 273 /**
 274  * Other malicious peers
 275  */
 276 static struct GNUNET_PeerIdentity *mal_peers;
 277
 278 /**
 279  * Hashmap of malicious peers used as set.
 280  * Used to more efficiently check whether we know that peer.
 281  */
 282 static struct GNUNET_CONTAINER_MultiPeerMap *mal_peer_set;
 283
 284 /**
 285  * Number of other malicious peers
 286  */
 287 static uint32_t num_mal_peers;
 288
 289
 290 /**
 291  * If type is 2 This struct is used to store the attacked peers in a DLL
 292  */
 293 struct AttackedPeer
 294 {
 295   /**
 296    * DLL
 297    */
 298   struct AttackedPeer *next;
 299   struct AttackedPeer *prev;
 300
 301   /**
 302    * PeerID
 303    */
 304   struct GNUNET_PeerIdentity peer_id;
 305 };
 306
 307 /**
 308  * If type is 2 this is the DLL of attacked peers
 309  */
 310 static struct AttackedPeer *att_peers_head;
 311 static struct AttackedPeer *att_peers_tail;
 312
 313 /**
 314  * This index is used to point to an attacked peer to
 315  * implement the round-robin-ish way to select attacked peers.
 316  */
 317 static struct AttackedPeer *att_peer_index;
 318
 319 /**
 320  * Hashmap of attacked peers used as set.
 321  * Used to more efficiently check whether we know that peer.
 322  */
 323 static struct GNUNET_CONTAINER_MultiPeerMap *att_peer_set;
 324
 325 /**
 326  * Number of attacked peers
 327  */
 328 static uint32_t num_attacked_peers;
 329
 330 /**
 331  * If type is 1 this is the attacked peer
 332  */
 333 static struct GNUNET_PeerIdentity attacked_peer;
 334
 335 /**
 336  * The limit of PUSHes we can send in one round.
 337  * This is an assumption of the Brahms protocol and either implemented
 338  * via proof of work
 339  * or
 340  * assumend to be the bandwidth limitation.
 341  */
 342 static uint32_t push_limit = 10000;
 343 #endif /* ENABLE_MALICIOUS */
 344
 345
 346 /***********************************************************************
 347  * /Globals
 348 ***********************************************************************/
 349
 350
 351 /***********************************************************************
 352  * Util functions
 353 ***********************************************************************/
 354
 355
 356 /**
 357  * Print peerlist to log.
 358  */
 359 static void
 360 print_peer_list (struct GNUNET_PeerIdentity *list,
 361                  unsigned int len)
 362 {
 363   unsigned int i;
 364
 365   LOG (GNUNET_ERROR_TYPE_DEBUG,
 366        "Printing peer list of length %u at %p:\n",
 367        len,
 368        list);
 369   for (i = 0 ; i < len ; i++)
 370   {
 371     LOG (GNUNET_ERROR_TYPE_DEBUG,
 372          "%u. peer: %s\n",
 373          i, GNUNET_i2s (&list[i]));
 374   }
 375 }
 376
 377
 378 /**
 379  * Remove peer from list.
 380  */
 381 static void
 382 rem_from_list (struct GNUNET_PeerIdentity **peer_list,
 383                unsigned int *list_size,
 384                const struct GNUNET_PeerIdentity *peer)
 385 {
 386   unsigned int i;
 387   struct GNUNET_PeerIdentity *tmp;
 388
 389   tmp = *peer_list;
 390
 391   LOG (GNUNET_ERROR_TYPE_DEBUG,
 392        "Removing peer %s from list at %p\n",
 393        GNUNET_i2s (peer),
 394        tmp);
 395
 396   for ( i = 0 ; i < *list_size ; i++ )
 397   {
 398     if (0 == GNUNET_CRYPTO_cmp_peer_identity (&tmp[i], peer))
 399     {
 400       if (i < *list_size -1)
 401       { /* Not at the last entry -- shift peers left */
 402         GNUNET_memcpy (&tmp[i], &tmp[i +1],
 403                 ((*list_size) - i -1) * sizeof (struct GNUNET_PeerIdentity));
 404       }
 405       /* Remove last entry (should be now useless PeerID) */
 406       GNUNET_array_grow (tmp, *list_size, (*list_size) -1);
 407     }
 408   }
 409   *peer_list = tmp;
 410 }
 411
 412
 413 /**
 414  * Sum all time relatives of an array.
 415  */
 416 static struct GNUNET_TIME_Relative
 417 T_relative_sum (const struct GNUNET_TIME_Relative *rel_array,
 418                 uint32_t arr_size)
 419 {
 420   struct GNUNET_TIME_Relative sum;
 421   uint32_t i;
 422
 423   sum = GNUNET_TIME_UNIT_ZERO;
 424   for ( i = 0 ; i < arr_size ; i++ )
 425   {
 426     sum = GNUNET_TIME_relative_add (sum, rel_array[i]);
 427   }
 428   return sum;
 429 }
 430
 431
 432 /**
 433  * Compute the average of given time relatives.
 434  */
 435 static struct GNUNET_TIME_Relative
 436 T_relative_avg (const struct GNUNET_TIME_Relative *rel_array,
 437                 uint32_t arr_size)
 438 {
 439   return GNUNET_TIME_relative_divide (T_relative_sum (rel_array,
 440                                                       arr_size),
 441                                       arr_size);
 442 }
 443
 444
 445 /**
 446  * Insert PeerID in #view
 447  *
 448  * Called once we know a peer is live.
 449  * Implements #PeerOp
 450  *
 451  * @return GNUNET_OK if peer was actually inserted
 452  *         GNUNET_NO if peer was not inserted
 453  */
 454 static void
 455 insert_in_view_op (void *cls,
 456                 const struct GNUNET_PeerIdentity *peer);
 457
 458 /**
 459  * Insert PeerID in #view
 460  *
 461  * Called once we know a peer is live.
 462  *
 463  * @return GNUNET_OK if peer was actually inserted
 464  *         GNUNET_NO if peer was not inserted
 465  */
 466 static int
 467 insert_in_view (const struct GNUNET_PeerIdentity *peer)
 468 {
 469   int online;
 470
 471   online = Peers_check_peer_flag (peer, Peers_ONLINE);
 472   if ( (GNUNET_NO == online) ||
 473        (GNUNET_SYSERR == online) ) /* peer is not even known */
 474   {
 475     (void) Peers_issue_peer_liveliness_check (peer);
 476     (void) Peers_schedule_operation (peer, insert_in_view_op);
 477     return GNUNET_NO;
 478   }
 479   /* Open channel towards peer to keep connection open */
 480   Peers_indicate_sending_intention (peer);
 481   return View_put (peer);
 482 }
 483
 484 /**
 485  * Put random peer from sampler into the view as history update.
 486  */
 487 static void
 488 hist_update (void *cls,
 489              struct GNUNET_PeerIdentity *ids,
 490              uint32_t num_peers)
 491 {
 492   unsigned int i;
 493
 494   for (i = 0; i < num_peers; i++)
 495   {
 496     (void) insert_in_view (&ids[i]);
 497     to_file (file_name_view_log,
 498              "+%s\t(hist)",
 499              GNUNET_i2s_full (ids));
 500   }
 501 }
 502
 503
 504 /**
 505  * Wrapper around #RPS_sampler_resize()
 506  *
 507  * If we do not have enough sampler elements, double current sampler size
 508  * If we have more than enough sampler elements, halv current sampler size
 509  */
 510 static void
 511 resize_wrapper (struct RPS_Sampler *sampler, uint32_t new_size)
 512 {
 513   unsigned int sampler_size;
 514
 515   // TODO statistics
 516   // TODO respect the min, max
 517   sampler_size = RPS_sampler_get_size (sampler);
 518   if (sampler_size > new_size * 4)
 519   { /* Shrinking */
 520     RPS_sampler_resize (sampler, sampler_size / 2);
 521   }
 522   else if (sampler_size < new_size)
 523   { /* Growing */
 524     RPS_sampler_resize (sampler, sampler_size * 2);
 525   }
 526   LOG (GNUNET_ERROR_TYPE_DEBUG, "sampler_size is now %u\n", sampler_size);
 527 }
 528
 529
 530 /**
 531  * Wrapper around #RPS_sampler_resize() resizing the client sampler
 532  */
 533 static void
 534 client_resize_wrapper ()
 535 {
 536   uint32_t bigger_size;
 537
 538   // TODO statistics
 539
 540   bigger_size = GNUNET_MAX (sampler_size_est_need, sampler_size_client_need);
 541
 542   // TODO respect the min, max
 543   resize_wrapper (client_sampler, bigger_size);
 544   LOG (GNUNET_ERROR_TYPE_DEBUG, "sampler_size_client is now %" PRIu32 "\n",
 545       bigger_size);
 546 }
 547
 548
 549 /**
 550  * Estimate request rate
 551  *
 552  * Called every time we receive a request from the client.
 553  */
 554 static void
 555 est_request_rate()
 556 {
 557   struct GNUNET_TIME_Relative max_round_duration;
 558
 559   if (request_deltas_size > req_counter)
 560     req_counter++;
 561   if ( 1 < req_counter)
 562   {
 563     /* Shift last request deltas to the right */
 564     memmove (&request_deltas[1],
 565         request_deltas,
 566         (req_counter - 1) * sizeof (struct GNUNET_TIME_Relative));
 567
 568     /* Add current delta to beginning */
 569     request_deltas[0] =
 570         GNUNET_TIME_absolute_get_difference (last_request,
 571                                              GNUNET_TIME_absolute_get ());
 572     request_rate = T_relative_avg (request_deltas, req_counter);
 573     request_rate = (request_rate.rel_value_us < 1) ?
 574       GNUNET_TIME_relative_get_unit_ () : request_rate;
 575
 576     /* Compute the duration a round will maximally take */
 577     max_round_duration =
 578         GNUNET_TIME_relative_add (round_interval,
 579                                   GNUNET_TIME_relative_divide (round_interval, 2));
 580
 581     /* Set the estimated size the sampler has to have to
 582      * satisfy the current client request rate */
 583     sampler_size_client_need =
 584         max_round_duration.rel_value_us / request_rate.rel_value_us;
 585
 586     /* Resize the sampler */
 587     client_resize_wrapper ();
 588   }
 589   last_request = GNUNET_TIME_absolute_get ();
 590 }
 591
 592
 593 /**
 594  * Add all peers in @a peer_array to @a peer_map used as set.
 595  *
 596  * @param peer_array array containing the peers
 597  * @param num_peers number of peers in @peer_array
 598  * @param peer_map the peermap to use as set
 599  */
 600 static void
 601 add_peer_array_to_set (const struct GNUNET_PeerIdentity *peer_array,
 602                        unsigned int num_peers,
 603                        struct GNUNET_CONTAINER_MultiPeerMap *peer_map)
 604 {
 605   unsigned int i;
 606   if (NULL == peer_map)
 607   {
 608     LOG (GNUNET_ERROR_TYPE_WARNING,
 609          "Trying to add peers to non-existing peermap.\n");
 610     return;
 611   }
 612
 613   for (i = 0; i < num_peers; i++)
 614   {
 615     GNUNET_CONTAINER_multipeermap_put (peer_map,
 616                                        &peer_array[i],
 617                                        NULL,
 618                                        GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_FAST);
 619   }
 620 }
 621
 622
 623 /**
 624  * Send a PULL REPLY to @a peer_id
 625  *
 626  * @param peer_id the peer to send the reply to.
 627  * @param peer_ids the peers to send to @a peer_id
 628  * @param num_peer_ids the number of peers to send to @a peer_id
 629  */
 630 static void
 631 send_pull_reply (const struct GNUNET_PeerIdentity *peer_id,
 632                  const struct GNUNET_PeerIdentity *peer_ids,
 633                  unsigned int num_peer_ids)
 634 {
 635   uint32_t send_size;
 636   struct GNUNET_MQ_Envelope *ev;
 637   struct GNUNET_RPS_P2P_PullReplyMessage *out_msg;
 638
 639   /* Compute actual size */
 640   send_size = sizeof (struct GNUNET_RPS_P2P_PullReplyMessage) +
 641               num_peer_ids * sizeof (struct GNUNET_PeerIdentity);
 642
 643   if (GNUNET_CONSTANTS_MAX_CADET_MESSAGE_SIZE < send_size)
 644     /* Compute number of peers to send
 645      * If too long, simply truncate */
 646     // TODO select random ones via permutation
 647     //      or even better: do good protocol design
 648     send_size =
 649       (GNUNET_CONSTANTS_MAX_CADET_MESSAGE_SIZE -
 650        sizeof (struct GNUNET_RPS_P2P_PullReplyMessage)) /
 651        sizeof (struct GNUNET_PeerIdentity);
 652   else
 653     send_size = num_peer_ids;
 654
 655   LOG (GNUNET_ERROR_TYPE_DEBUG,
 656       "Going to send PULL REPLY with %u peers to %s\n",
 657       send_size, GNUNET_i2s (peer_id));
 658
 659   ev = GNUNET_MQ_msg_extra (out_msg,
 660                             send_size * sizeof (struct GNUNET_PeerIdentity),
 661                             GNUNET_MESSAGE_TYPE_RPS_PP_PULL_REPLY);
 662   out_msg->num_peers = htonl (send_size);
 663   GNUNET_memcpy (&out_msg[1], peer_ids,
 664          send_size * sizeof (struct GNUNET_PeerIdentity));
 665
 666   Peers_send_message (peer_id, ev, "PULL REPLY");
 667 }
 668
 669
 670 /**
 671  * Insert PeerID in #pull_map
 672  *
 673  * Called once we know a peer is live.
 674  */
 675 static void
 676 insert_in_pull_map (void *cls,
 677                     const struct GNUNET_PeerIdentity *peer)
 678 {
 679   CustomPeerMap_put (pull_map, peer);
 680 }
 681
 682
 683 /**
 684  * Insert PeerID in #view
 685  *
 686  * Called once we know a peer is live.
 687  * Implements #PeerOp
 688  */
 689 static void
 690 insert_in_view_op (void *cls,
 691                 const struct GNUNET_PeerIdentity *peer)
 692 {
 693   (void) insert_in_view (peer);
 694 }
 695
 696
 697 /**
 698  * Update sampler with given PeerID.
 699  * Implements #PeerOp
 700  */
 701 static void
 702 insert_in_sampler (void *cls,
 703                    const struct GNUNET_PeerIdentity *peer)
 704 {
 705   LOG (GNUNET_ERROR_TYPE_DEBUG,
 706        "Updating samplers with peer %s from insert_in_sampler()\n",
 707        GNUNET_i2s (peer));
 708   RPS_sampler_update (prot_sampler,   peer);
 709   RPS_sampler_update (client_sampler, peer);
 710   if (0 < RPS_sampler_count_id (prot_sampler, peer))
 711   {
 712     /* Make sure we 'know' about this peer */
 713     (void) Peers_issue_peer_liveliness_check (peer);
 714     /* Establish a channel towards that peer to indicate we are going to send
 715      * messages to it */
 716     //Peers_indicate_sending_intention (peer);
 717   }
 718 }
 719
 720 /**
 721  * @brief This is called on peers from external sources (cadet, peerinfo, ...)
 722  *        If the peer is not known, liveliness check is issued and it is
 723  *        scheduled to be inserted in sampler and view.
 724  *
 725  * "External sources" refer to every source except the gossip.
 726  *
 727  * @param peer peer to insert
 728  */
 729 static void
 730 got_peer (const struct GNUNET_PeerIdentity *peer)
 731 {
 732   /* If we did not know this peer already, insert it into sampler and view */
 733   if (GNUNET_YES == Peers_issue_peer_liveliness_check (peer))
 734   {
 735     Peers_schedule_operation (peer, insert_in_sampler);
 736     Peers_schedule_operation (peer, insert_in_view_op);
 737   }
 738 }
 739
 740 /**
 741  * @brief Checks if there is a sending channel and if it is needed
 742  *
 743  * @param peer the peer whose sending channel is checked
 744  * @return GNUNET_YES if sending channel exists and is still needed
 745  *         GNUNET_NO  otherwise
 746  */
 747 static int
 748 check_sending_channel_needed (const struct GNUNET_PeerIdentity *peer)
 749 {
 750   /* struct GNUNET_CADET_Channel *channel; */
 751   if (GNUNET_NO == Peers_check_peer_known (peer))
 752   {
 753     return GNUNET_NO;
 754   }
 755   if (GNUNET_YES == Peers_check_sending_channel_exists (peer))
 756   {
 757     if ( (0 < RPS_sampler_count_id (prot_sampler, peer)) ||
 758          (GNUNET_YES == View_contains_peer (peer)) ||
 759          (GNUNET_YES == CustomPeerMap_contains_peer (push_map, peer)) ||
 760          (GNUNET_YES == CustomPeerMap_contains_peer (pull_map, peer)) ||
 761          (GNUNET_YES == Peers_check_peer_flag (peer, Peers_PULL_REPLY_PENDING)))
 762     { /* If we want to keep the connection to peer open */
 763       return GNUNET_YES;
 764     }
 765     return GNUNET_NO;
 766   }
 767   return GNUNET_NO;
 768 }
 769
 770 /**
 771  * @brief remove peer from our knowledge, the view, push and pull maps and
 772  * samplers.
 773  *
 774  * @param peer the peer to remove
 775  */
 776 static void
 777 remove_peer (const struct GNUNET_PeerIdentity *peer)
 778 {
 779   (void) View_remove_peer (peer);
 780   CustomPeerMap_remove_peer (pull_map, peer);
 781   CustomPeerMap_remove_peer (push_map, peer);
 782   RPS_sampler_reinitialise_by_value (prot_sampler, peer);
 783   RPS_sampler_reinitialise_by_value (client_sampler, peer);
 784   Peers_remove_peer (peer);
 785 }
 786
 787
 788 /**
 789  * @brief Remove data that is not needed anymore.
 790  *
 791  * If the sending channel is no longer needed it is destroyed.
 792  *
 793  * @param peer the peer whose data is about to be cleaned
 794  */
 795 static void
 796 clean_peer (const struct GNUNET_PeerIdentity *peer)
 797 {
 798   if (GNUNET_NO == check_sending_channel_needed (peer))
 799   {
 800     LOG (GNUNET_ERROR_TYPE_DEBUG,
 801         "Going to remove send channel to peer %s\n",
 802         GNUNET_i2s (peer));
 803     #ifdef ENABLE_MALICIOUS
 804     if (0 != GNUNET_CRYPTO_cmp_peer_identity (&attacked_peer, peer))
 805       (void) Peers_destroy_sending_channel (peer);
 806     #else /* ENABLE_MALICIOUS */
 807     (void) Peers_destroy_sending_channel (peer);
 808     #endif /* ENABLE_MALICIOUS */
 809   }
 810
 811   if ( (GNUNET_NO == Peers_check_peer_send_intention (peer)) &&
 812        (GNUNET_NO == View_contains_peer (peer)) &&
 813        (GNUNET_NO == CustomPeerMap_contains_peer (push_map, peer)) &&
 814        (GNUNET_NO == CustomPeerMap_contains_peer (push_map, peer)) &&
 815        (0 == RPS_sampler_count_id (prot_sampler,   peer)) &&
 816        (0 == RPS_sampler_count_id (client_sampler, peer)) &&
 817        (GNUNET_NO != Peers_check_removable (peer)) )
 818   { /* We can safely remove this peer */
 819     LOG (GNUNET_ERROR_TYPE_DEBUG,
 820         "Going to remove peer %s\n",
 821         GNUNET_i2s (peer));
 822     remove_peer (peer);
 823     return;
 824   }
 825 }
 826
 827 /**
 828  * @brief This is called when a channel is destroyed.
 829  *
 830  * Removes peer completely from our knowledge if the send_channel was destroyed
 831  * Otherwise simply delete the recv_channel
 832  * Also check if the knowledge about this peer is still needed.
 833  * If not, remove this peer from our knowledge.
 834  *
 835  * @param cls The closure
 836  * @param channel The channel being closed
 837  * @param channel_ctx The context associated with this channel
 838  */
 839 static void
 840 cleanup_destroyed_channel (void *cls,
 841                            const struct GNUNET_CADET_Channel *channel,
 842                            void *channel_ctx)
 843 {
 844   struct GNUNET_PeerIdentity *peer;
 845
 846   peer = (struct GNUNET_PeerIdentity *) GNUNET_CADET_channel_get_info (
 847       (struct GNUNET_CADET_Channel *) channel, GNUNET_CADET_OPTION_PEER);
 848        // FIXME wait for cadet to change this function
 849
 850   if (GNUNET_NO == Peers_check_peer_known (peer))
 851   { /* We don't know a context to that peer */
 852     LOG (GNUNET_ERROR_TYPE_WARNING,
 853          "channel (%s) without associated context was destroyed\n",
 854          GNUNET_i2s (peer));
 855     return;
 856   }
 857
 858   if (GNUNET_YES == Peers_check_peer_flag (peer, Peers_TO_DESTROY))
 859   { /* We are in the middle of removing that peer from our knowledge. In this
 860        case simply make sure that the channels are cleaned. */
 861     Peers_cleanup_destroyed_channel (cls, channel, channel_ctx);
 862     to_file (file_name_view_log,
 863              "-%s\t(cleanup channel, ourself)",
 864              GNUNET_i2s_full (peer));
 865     return;
 866   }
 867
 868   if (GNUNET_YES ==
 869       Peers_check_channel_role (peer, channel, Peers_CHANNEL_ROLE_SENDING))
 870   { /* Channel used for sending was destroyed */
 871     /* Possible causes of channel destruction:
 872      *  - ourselves  -> cleaning send channel -> clean context
 873      *  - other peer -> peer probably went down -> remove
 874      */
 875     if (GNUNET_YES == Peers_check_channel_flag (channel_ctx, Peers_CHANNEL_CLEAN))
 876     { /* We are about to clean the sending channel. Clean the respective
 877        * context */
 878       Peers_cleanup_destroyed_channel (cls, channel, channel_ctx);
 879       return;
 880     }
 881     else
 882     { /* Other peer destroyed our sending channel that he is supposed to keep
 883        * open. It probably went down. Remove it from our knowledge. */
 884       Peers_cleanup_destroyed_channel (cls, channel, channel_ctx);
 885       remove_peer (peer);
 886       return;
 887     }
 888   }
 889   else if (GNUNET_YES ==
 890       Peers_check_channel_role (peer, channel, Peers_CHANNEL_ROLE_RECEIVING))
 891   { /* Channel used for receiving was destroyed */
 892     /* Possible causes of channel destruction:
 893      *  - ourselves  -> peer tried to establish channel twice -> clean context
 894      *  - other peer -> peer doesn't want to send us data -> clean
 895      */
 896     if (GNUNET_YES ==
 897         Peers_check_channel_flag (channel_ctx, Peers_CHANNEL_ESTABLISHED_TWICE))
 898     { /* Other peer tried to establish a channel to us twice. We do not accept
 899        * that. Clean the context. */
 900       Peers_cleanup_destroyed_channel (cls, channel, channel_ctx);
 901       return;
 902     }
 903     else
 904     { /* Other peer doesn't want to send us data anymore. We are free to clean
 905        * it. */
 906       Peers_cleanup_destroyed_channel (cls, channel, channel_ctx);
 907       clean_peer (peer);
 908       return;
 909     }
 910   }
 911   else
 912   {
 913     LOG (GNUNET_ERROR_TYPE_WARNING,
 914         "Destroyed channel is neither sending nor receiving channel\n");
 915   }
 916 }
 917
 918 /***********************************************************************
 919  * /Util functions
 920 ***********************************************************************/
 921
 922 static void
 923 destroy_reply_cls (struct ReplyCls *rep_cls)
 924 {
 925   struct ClientContext *cli_ctx;
 926
 927   cli_ctx = rep_cls->cli_ctx;
 928   GNUNET_assert (NULL != cli_ctx);
 929   RPS_sampler_request_cancel (rep_cls->req_handle);
 930   GNUNET_CONTAINER_DLL_remove (cli_ctx->rep_cls_head,
 931                                cli_ctx->rep_cls_tail,
 932                                rep_cls);
 933   GNUNET_free (rep_cls);
 934 }
 935
 936
 937 static void
 938 destroy_cli_ctx (struct ClientContext *cli_ctx)
 939 {
 940   GNUNET_assert (NULL != cli_ctx);
 941   if (NULL != cli_ctx->rep_cls_head)
 942   {
 943     LOG (GNUNET_ERROR_TYPE_WARNING,
 944          "Trying to destroy the context of a client that still has pending requests. Going to clean those\n");
 945     while (NULL != cli_ctx->rep_cls_head)
 946       destroy_reply_cls (cli_ctx->rep_cls_head);
 947   }
 948   GNUNET_CONTAINER_DLL_remove (cli_ctx_head,
 949                                cli_ctx_tail,
 950                                cli_ctx);
 951   GNUNET_free (cli_ctx);
 952 }
 953
 954
 955 /**
 956  * Function called by NSE.
 957  *
 958  * Updates sizes of sampler list and view and adapt those lists
 959  * accordingly.
 960  */
 961 static void
 962 nse_callback (void *cls,
 963               struct GNUNET_TIME_Absolute timestamp,
 964               double logestimate, double std_dev)
 965 {
 966   double estimate;
 967   //double scale; // TODO this might go gloabal/config
 968
 969   LOG (GNUNET_ERROR_TYPE_DEBUG,
 970        "Received a ns estimate - logest: %f, std_dev: %f (old_size: %u)\n",
 971        logestimate, std_dev, RPS_sampler_get_size (prot_sampler));
 972   //scale = .01;
 973   estimate = GNUNET_NSE_log_estimate_to_n (logestimate);
 974   // GNUNET_NSE_log_estimate_to_n (logestimate);
 975   estimate = pow (estimate, 1.0 / 3);
 976   // TODO add if std_dev is a number
 977   // estimate += (std_dev * scale);
 978   if (2 < ceil (estimate))
 979   {
 980     LOG (GNUNET_ERROR_TYPE_DEBUG, "Changing estimate to %f\n", estimate);
 981     sampler_size_est_need = estimate;
 982   } else
 983     LOG (GNUNET_ERROR_TYPE_DEBUG, "Not using estimate %f\n", estimate);
 984
 985   /* If the NSE has changed adapt the lists accordingly */
 986   resize_wrapper (prot_sampler, sampler_size_est_need);
 987   client_resize_wrapper ();
 988 }
 989
 990
 991 /**
 992  * Callback called once the requested PeerIDs are ready.
 993  *
 994  * Sends those to the requesting client.
 995  */
 996 static void
 997 client_respond (void *cls,
 998                 struct GNUNET_PeerIdentity *peer_ids,
 999                 uint32_t num_peers)
1000 {
1001   struct ReplyCls *reply_cls = cls;
1002   uint32_t i;
1003   struct GNUNET_MQ_Envelope *ev;
1004   struct GNUNET_RPS_CS_ReplyMessage *out_msg;
1005   uint32_t size_needed;
1006   struct ClientContext *cli_ctx;
1007
1008   GNUNET_assert (NULL != reply_cls);
1009   LOG (GNUNET_ERROR_TYPE_DEBUG,
1010        "sampler returned %" PRIu32 " peers:\n",
1011        num_peers);
1012   for (i = 0; i < num_peers; i++)
1013   {
1014     LOG (GNUNET_ERROR_TYPE_DEBUG,
1015          "  %" PRIu32 ": %s\n",
1016          i,
1017          GNUNET_i2s (&peer_ids[i]));
1018   }
1019
1020   size_needed = sizeof (struct GNUNET_RPS_CS_ReplyMessage) +
1021                 num_peers * sizeof (struct GNUNET_PeerIdentity);
1022
1023   GNUNET_assert (GNUNET_SERVER_MAX_MESSAGE_SIZE >= size_needed);
1024
1025   ev = GNUNET_MQ_msg_extra (out_msg,
1026                             num_peers * sizeof (struct GNUNET_PeerIdentity),
1027                             GNUNET_MESSAGE_TYPE_RPS_CS_REPLY);
1028   out_msg->num_peers = htonl (num_peers);
1029   out_msg->id = htonl (reply_cls->id);
1030
1031   GNUNET_memcpy (&out_msg[1],
1032           peer_ids,
1033           num_peers * sizeof (struct GNUNET_PeerIdentity));
1034   GNUNET_free (peer_ids);
1035
1036   cli_ctx = reply_cls->cli_ctx;
1037   GNUNET_assert (NULL != cli_ctx);
1038   destroy_reply_cls (reply_cls);
1039   GNUNET_MQ_send (cli_ctx->mq, ev);
1040 }
1041
1042
1043 /**
1044  * Handle RPS request from the client.
1045  *
1046  * @param cls closure
1047  * @param client identification of the client
1048  * @param message the actual message
1049  */
1050 static void
1051 handle_client_request (void *cls,
1052                        const struct GNUNET_RPS_CS_RequestMessage *msg)
1053 {
1054   struct ClientContext *cli_ctx = cls;
1055   uint32_t num_peers;
1056   uint32_t size_needed;
1057   struct ReplyCls *reply_cls;
1058   uint32_t i;
1059
1060   num_peers = ntohl (msg->num_peers);
1061   size_needed = sizeof (struct GNUNET_RPS_CS_RequestMessage) +
1062                 num_peers * sizeof (struct GNUNET_PeerIdentity);
1063
1064   if (GNUNET_SERVER_MAX_MESSAGE_SIZE < size_needed)
1065   {
1066     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
1067                 "Message received from client has size larger than expected\n");
1068     GNUNET_SERVICE_client_drop (cli_ctx->client);
1069     return;
1070   }
1071
1072   for (i = 0 ; i < num_peers ; i++)
1073     est_request_rate();
1074
1075   LOG (GNUNET_ERROR_TYPE_DEBUG,
1076        "Client requested %" PRIu32 " random peer(s).\n",
1077        num_peers);
1078
1079   reply_cls = GNUNET_new (struct ReplyCls);
1080   reply_cls->id = ntohl (msg->id);
1081   reply_cls->cli_ctx = cli_ctx;
1082   reply_cls->req_handle = RPS_sampler_get_n_rand_peers (client_sampler,
1083                                                         client_respond,
1084                                                         reply_cls,
1085                                                         num_peers);
1086
1087   GNUNET_assert (NULL != cli_ctx);
1088   GNUNET_CONTAINER_DLL_insert (cli_ctx->rep_cls_head,
1089                                cli_ctx->rep_cls_tail,
1090                                reply_cls);
1091   GNUNET_SERVICE_client_continue (cli_ctx->client);
1092 }
1093
1094
1095 /**
1096  * @brief Handle a message that requests the cancellation of a request
1097  *
1098  * @param cls unused
1099  * @param client the client that requests the cancellation
1100  * @param message the message containing the id of the request
1101  */
1102 static void
1103 handle_client_request_cancel (void *cls,
1104                           const struct GNUNET_RPS_CS_RequestCancelMessage *msg)
1105 {
1106   struct ClientContext *cli_ctx = cls;
1107   struct ReplyCls *rep_cls;
1108
1109   GNUNET_assert (NULL != cli_ctx);
1110   GNUNET_assert (NULL != cli_ctx->rep_cls_head);
1111   rep_cls = cli_ctx->rep_cls_head;
1112   LOG (GNUNET_ERROR_TYPE_DEBUG,
1113       "Client cancels request with id %" PRIu32 "\n",
1114       ntohl (msg->id));
1115   while ( (NULL != rep_cls->next) &&
1116           (rep_cls->id != ntohl (msg->id)) )
1117     rep_cls = rep_cls->next;
1118   GNUNET_assert (rep_cls->id == ntohl (msg->id));
1119   RPS_sampler_request_cancel (rep_cls->req_handle);
1120   destroy_reply_cls (rep_cls);
1121   GNUNET_SERVICE_client_continue (cli_ctx->client);
1122 }
1123
1124
1125 /**
1126  * Handle seed from the client.
1127  *
1128  * @param cls closure
1129  * @param client identification of the client
1130  * @param message the actual message
1131  */
1132 static void
1133 handle_client_seed (void *cls,
1134                     const struct GNUNET_MessageHeader *message)
1135 {
1136   struct ClientContext *cli_ctx = cls;
1137   struct GNUNET_RPS_CS_SeedMessage *in_msg;
1138   struct GNUNET_PeerIdentity *peers;
1139   uint32_t num_peers;
1140   uint32_t i;
1141
1142   if (sizeof (struct GNUNET_RPS_CS_SeedMessage) > ntohs (message->size))
1143   {
1144     GNUNET_SERVICE_client_drop (cli_ctx->client);
1145     GNUNET_break_op (0);
1146     return;
1147   }
1148
1149   in_msg = (struct GNUNET_RPS_CS_SeedMessage *) message;
1150   num_peers = ntohl (in_msg->num_peers);
1151   peers = (struct GNUNET_PeerIdentity *) &in_msg[1];
1152   //peers = GNUNET_new_array (num_peers, struct GNUNET_PeerIdentity);
1153   //GNUNET_memcpy (peers, &in_msg[1], num_peers * sizeof (struct GNUNET_PeerIdentity));
1154
1155   if ((ntohs (message->size) - sizeof (struct GNUNET_RPS_CS_SeedMessage)) /
1156       sizeof (struct GNUNET_PeerIdentity) != num_peers)
1157   {
1158     GNUNET_break_op (0);
1159     GNUNET_SERVICE_client_drop (cli_ctx->client);
1160     return;
1161   }
1162
1163   LOG (GNUNET_ERROR_TYPE_DEBUG,
1164        "Client seeded peers:\n");
1165   print_peer_list (peers, num_peers);
1166
1167   for (i = 0; i < num_peers; i++)
1168   {
1169     LOG (GNUNET_ERROR_TYPE_DEBUG,
1170          "Updating samplers with seed %" PRIu32 ": %s\n",
1171          i,
1172          GNUNET_i2s (&peers[i]));
1173
1174     got_peer (&peers[i]);
1175
1176     //RPS_sampler_update (prot_sampler,   &peers[i]);
1177     //RPS_sampler_update (client_sampler, &peers[i]);
1178   }
1179
1180   ////GNUNET_free (peers);
1181
1182   GNUNET_SERVICE_client_drop (cli_ctx->client);
1183 }
1184
1185 /**
1186  * Handle a CHECK_LIVE message from another peer.
1187  *
1188  * This does nothing. But without calling #GNUNET_CADET_receive_done()
1189  * the channel is blocked for all other communication.
1190  *
1191  * @param cls Closure
1192  * @param channel The channel the CHECK was received over
1193  * @param channel_ctx The context associated with this channel
1194  * @param msg The message header
1195  */
1196 static int
1197 handle_peer_check (void *cls,
1198                   struct GNUNET_CADET_Channel *channel,
1199                   void **channel_ctx,
1200                   const struct GNUNET_MessageHeader *msg)
1201 {
1202   GNUNET_CADET_receive_done (channel);
1203   return GNUNET_OK;
1204 }
1205
1206 /**
1207  * Handle a PUSH message from another peer.
1208  *
1209  * Check the proof of work and store the PeerID
1210  * in the temporary list for pushed PeerIDs.
1211  *
1212  * @param cls Closure
1213  * @param channel The channel the PUSH was received over
1214  * @param channel_ctx The context associated with this channel
1215  * @param msg The message header
1216  */
1217 static int
1218 handle_peer_push (void *cls,
1219                   struct GNUNET_CADET_Channel *channel,
1220                   void **channel_ctx,
1221                   const struct GNUNET_MessageHeader *msg)
1222 {
1223   const struct GNUNET_PeerIdentity *peer;
1224
1225   // (check the proof of work (?))
1226
1227   peer = (const struct GNUNET_PeerIdentity *)
1228     GNUNET_CADET_channel_get_info (channel, GNUNET_CADET_OPTION_PEER);
1229   // FIXME wait for cadet to change this function
1230
1231   LOG (GNUNET_ERROR_TYPE_DEBUG,
1232        "Received PUSH (%s)\n",
1233        GNUNET_i2s (peer));
1234
1235   #ifdef ENABLE_MALICIOUS
1236   struct AttackedPeer *tmp_att_peer;
1237
1238   if ( (1 == mal_type) ||
1239        (3 == mal_type) )
1240   { /* Try to maximise representation */
1241     tmp_att_peer = GNUNET_new (struct AttackedPeer);
1242     GNUNET_memcpy (&tmp_att_peer->peer_id, peer, sizeof (struct GNUNET_PeerIdentity));
1243     if (NULL == att_peer_set)
1244       att_peer_set = GNUNET_CONTAINER_multipeermap_create (1, GNUNET_NO);
1245     if (GNUNET_NO == GNUNET_CONTAINER_multipeermap_contains (att_peer_set,
1246                                                              peer))
1247     {
1248       GNUNET_CONTAINER_DLL_insert (att_peers_head,
1249                                    att_peers_tail,
1250                                    tmp_att_peer);
1251       add_peer_array_to_set (peer, 1, att_peer_set);
1252     }
1253     GNUNET_CADET_receive_done (channel);
1254     return GNUNET_OK;
1255   }
1256
1257
1258   else if (2 == mal_type)
1259   { /* We attack one single well-known peer - simply ignore */
1260     GNUNET_CADET_receive_done (channel);
1261     return GNUNET_OK;
1262   }
1263   #endif /* ENABLE_MALICIOUS */
1264
1265   /* Add the sending peer to the push_map */
1266   CustomPeerMap_put (push_map, peer);
1267
1268   GNUNET_CADET_receive_done (channel);
1269   return GNUNET_OK;
1270 }
1271
1272
1273 /**
1274  * Handle PULL REQUEST request message from another peer.
1275  *
1276  * Reply with the view of PeerIDs.
1277  *
1278  * @param cls Closure
1279  * @param channel The channel the PULL REQUEST was received over
1280  * @param channel_ctx The context associated with this channel
1281  * @param msg The message header
1282  */
1283 static int
1284 handle_peer_pull_request (void *cls,
1285                           struct GNUNET_CADET_Channel *channel,
1286                           void **channel_ctx,
1287                           const struct GNUNET_MessageHeader *msg)
1288 {
1289   struct GNUNET_PeerIdentity *peer;
1290   const struct GNUNET_PeerIdentity *view_array;
1291
1292   peer = (struct GNUNET_PeerIdentity *)
1293     GNUNET_CADET_channel_get_info (channel,
1294                                    GNUNET_CADET_OPTION_PEER);
1295   // FIXME wait for cadet to change this function
1296
1297   LOG (GNUNET_ERROR_TYPE_DEBUG, "Received PULL REQUEST (%s)\n", GNUNET_i2s (peer));
1298
1299   #ifdef ENABLE_MALICIOUS
1300   if (1 == mal_type
1301       || 3 == mal_type)
1302   { /* Try to maximise representation */
1303     send_pull_reply (peer, mal_peers, num_mal_peers);
1304     GNUNET_CADET_receive_done (channel);
1305     return GNUNET_OK;
1306   }
1307
1308   else if (2 == mal_type)
1309   { /* Try to partition network */
1310     if (0 == GNUNET_CRYPTO_cmp_peer_identity (&attacked_peer, peer))
1311     {
1312       send_pull_reply (peer, mal_peers, num_mal_peers);
1313     }
1314     GNUNET_CADET_receive_done (channel);
1315     return GNUNET_OK;
1316   }
1317   #endif /* ENABLE_MALICIOUS */
1318
1319   view_array = View_get_as_array ();
1320
1321   send_pull_reply (peer, view_array, View_size ());
1322
1323   GNUNET_CADET_receive_done (channel);
1324   return GNUNET_OK;
1325 }
1326
1327
1328 /**
1329  * Handle PULL REPLY message from another peer.
1330  *
1331  * Check whether we sent a corresponding request and
1332  * whether this reply is the first one.
1333  *
1334  * @param cls Closure
1335  * @param channel The channel the PUSH was received over
1336  * @param channel_ctx The context associated with this channel
1337  * @param msg The message header
1338  */
1339 static int
1340 handle_peer_pull_reply (void *cls,
1341                         struct GNUNET_CADET_Channel *channel,
1342                         void **channel_ctx,
1343                         const struct GNUNET_MessageHeader *msg)
1344 {
1345   struct GNUNET_RPS_P2P_PullReplyMessage *in_msg;
1346   struct GNUNET_PeerIdentity *peers;
1347   struct GNUNET_PeerIdentity *sender;
1348   uint32_t i;
1349 #ifdef ENABLE_MALICIOUS
1350   struct AttackedPeer *tmp_att_peer;
1351 #endif /* ENABLE_MALICIOUS */
1352
1353   /* Check for protocol violation */
1354   if (sizeof (struct GNUNET_RPS_P2P_PullReplyMessage) > ntohs (msg->size))
1355   {
1356     GNUNET_break_op (0);
1357     GNUNET_CADET_receive_done (channel);
1358     return GNUNET_SYSERR;
1359   }
1360
1361   in_msg = (struct GNUNET_RPS_P2P_PullReplyMessage *) msg;
1362   if ((ntohs (msg->size) - sizeof (struct GNUNET_RPS_P2P_PullReplyMessage)) /
1363       sizeof (struct GNUNET_PeerIdentity) != ntohl (in_msg->num_peers))
1364   {
1365     LOG (GNUNET_ERROR_TYPE_ERROR,
1366         "message says it sends %" PRIu32 " peers, have space for %lu peers\n",
1367         ntohl (in_msg->num_peers),
1368         (ntohs (msg->size) - sizeof (struct GNUNET_RPS_P2P_PullReplyMessage)) /
1369             sizeof (struct GNUNET_PeerIdentity));
1370     GNUNET_break_op (0);
1371     GNUNET_CADET_receive_done (channel);
1372     return GNUNET_SYSERR;
1373   }
1374
1375   // Guess simply casting isn't the nicest way...
1376   // FIXME wait for cadet to change this function
1377   sender = (struct GNUNET_PeerIdentity *)
1378       GNUNET_CADET_channel_get_info (channel, GNUNET_CADET_OPTION_PEER);
1379
1380   LOG (GNUNET_ERROR_TYPE_DEBUG, "Received PULL REPLY (%s)\n", GNUNET_i2s (sender));
1381
1382   if (GNUNET_YES != Peers_check_peer_flag (sender, Peers_PULL_REPLY_PENDING))
1383   {
1384     LOG (GNUNET_ERROR_TYPE_WARNING,
1385         "Received a pull reply from a peer we didn't request one from!\n");
1386     GNUNET_CADET_receive_done (channel);
1387     GNUNET_break_op (0);
1388     return GNUNET_OK;
1389   }
1390
1391
1392   #ifdef ENABLE_MALICIOUS
1393   // We shouldn't even receive pull replies as we're not sending
1394   if (2 == mal_type)
1395   {
1396     GNUNET_CADET_receive_done (channel);
1397     return GNUNET_OK;
1398   }
1399   #endif /* ENABLE_MALICIOUS */
1400
1401   /* Do actual logic */
1402   peers = (struct GNUNET_PeerIdentity *) &in_msg[1];
1403
1404   LOG (GNUNET_ERROR_TYPE_DEBUG,
1405        "PULL REPLY received, got following %u peers:\n",
1406        ntohl (in_msg->num_peers));
1407
1408   for (i = 0 ; i < ntohl (in_msg->num_peers) ; i++)
1409   {
1410     LOG (GNUNET_ERROR_TYPE_DEBUG,
1411          "%u. %s\n",
1412          i,
1413          GNUNET_i2s (&peers[i]));
1414
1415     #ifdef ENABLE_MALICIOUS
1416     if ((NULL != att_peer_set) &&
1417         (1 == mal_type || 3 == mal_type))
1418     { /* Add attacked peer to local list */
1419       // TODO check if we sent a request and this was the first reply
1420       if (GNUNET_NO == GNUNET_CONTAINER_multipeermap_contains (att_peer_set,
1421                                                                &peers[i])
1422           && GNUNET_NO == GNUNET_CONTAINER_multipeermap_contains (mal_peer_set,
1423                                                                   &peers[i])
1424           && 0 != GNUNET_CRYPTO_cmp_peer_identity (&peers[i],
1425                                                    &own_identity))
1426       {
1427         tmp_att_peer = GNUNET_new (struct AttackedPeer);
1428         tmp_att_peer->peer_id = peers[i];
1429         GNUNET_CONTAINER_DLL_insert (att_peers_head,
1430                                      att_peers_tail,
1431                                      tmp_att_peer);
1432         add_peer_array_to_set (&peers[i], 1, att_peer_set);
1433       }
1434       continue;
1435     }
1436     #endif /* ENABLE_MALICIOUS */
1437     if (0 != GNUNET_CRYPTO_cmp_peer_identity (&own_identity,
1438                                               &peers[i]))
1439     {
1440       /* Make sure we 'know' about this peer */
1441       (void) Peers_insert_peer (&peers[i]);
1442
1443       if (GNUNET_YES == Peers_check_peer_valid (&peers[i]))
1444       {
1445         CustomPeerMap_put (pull_map, &peers[i]);
1446       }
1447       else
1448       {
1449         Peers_schedule_operation (&peers[i], insert_in_pull_map);
1450         (void) Peers_issue_peer_liveliness_check (&peers[i]);
1451       }
1452     }
1453   }
1454
1455   Peers_unset_peer_flag (sender, Peers_PULL_REPLY_PENDING);
1456   clean_peer (sender);
1457
1458   GNUNET_CADET_receive_done (channel);
1459   return GNUNET_OK;
1460 }
1461
1462
1463 /**
1464  * Compute a random delay.
1465  * A uniformly distributed value between mean + spread and mean - spread.
1466  *
1467  * For example for mean 4 min and spread 2 the minimum is (4 min - (1/2 * 4 min))
1468  * It would return a random value between 2 and 6 min.
1469  *
1470  * @param mean the mean
1471  * @param spread the inverse amount of deviation from the mean
1472  */
1473 static struct GNUNET_TIME_Relative
1474 compute_rand_delay (struct GNUNET_TIME_Relative mean,
1475                     unsigned int spread)
1476 {
1477   struct GNUNET_TIME_Relative half_interval;
1478   struct GNUNET_TIME_Relative ret;
1479   unsigned int rand_delay;
1480   unsigned int max_rand_delay;
1481
1482   if (0 == spread)
1483   {
1484     LOG (GNUNET_ERROR_TYPE_WARNING,
1485          "Not accepting spread of 0\n");
1486     GNUNET_break (0);
1487     GNUNET_assert (0);
1488   }
1489   GNUNET_assert (0 != mean.rel_value_us);
1490
1491   /* Compute random time value between spread * mean and spread * mean */
1492   half_interval = GNUNET_TIME_relative_divide (mean, spread);
1493
1494   max_rand_delay = GNUNET_TIME_UNIT_FOREVER_REL.rel_value_us / mean.rel_value_us * (2/spread);
1495   /**
1496    * Compute random value between (0 and 1) * round_interval
1497    * via multiplying round_interval with a 'fraction' (0 to value)/value
1498    */
1499   rand_delay = GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK, max_rand_delay);
1500   ret = GNUNET_TIME_relative_multiply (mean,  rand_delay);
1501   ret = GNUNET_TIME_relative_divide   (ret, max_rand_delay);
1502   ret = GNUNET_TIME_relative_add      (ret, half_interval);
1503
1504   if (GNUNET_TIME_UNIT_FOREVER_REL.rel_value_us == ret.rel_value_us)
1505     LOG (GNUNET_ERROR_TYPE_WARNING,
1506          "Returning FOREVER_REL\n");
1507
1508   return ret;
1509 }
1510
1511
1512 /**
1513  * Send single pull request
1514  *
1515  * @param peer_id the peer to send the pull request to.
1516  */
1517 static void
1518 send_pull_request (const struct GNUNET_PeerIdentity *peer)
1519 {
1520   struct GNUNET_MQ_Envelope *ev;
1521
1522   GNUNET_assert (GNUNET_NO == Peers_check_peer_flag (peer,
1523                                                      Peers_PULL_REPLY_PENDING));
1524   Peers_set_peer_flag (peer, Peers_PULL_REPLY_PENDING);
1525
1526   LOG (GNUNET_ERROR_TYPE_DEBUG,
1527        "Going to send PULL REQUEST to peer %s.\n",
1528        GNUNET_i2s (peer));
1529
1530   ev = GNUNET_MQ_msg_header (GNUNET_MESSAGE_TYPE_RPS_PP_PULL_REQUEST);
1531   Peers_send_message (peer, ev, "PULL REQUEST");
1532 }
1533
1534
1535 /**
1536  * Send single push
1537  *
1538  * @param peer_id the peer to send the push to.
1539  */
1540 static void
1541 send_push (const struct GNUNET_PeerIdentity *peer_id)
1542 {
1543   struct GNUNET_MQ_Envelope *ev;
1544
1545   LOG (GNUNET_ERROR_TYPE_DEBUG,
1546        "Going to send PUSH to peer %s.\n",
1547        GNUNET_i2s (peer_id));
1548
1549   ev = GNUNET_MQ_msg_header (GNUNET_MESSAGE_TYPE_RPS_PP_PUSH);
1550   Peers_send_message (peer_id, ev, "PUSH");
1551 }
1552
1553
1554 static void
1555 do_round (void *cls);
1556
1557 static void
1558 do_mal_round (void *cls);
1559
1560
1561 #ifdef ENABLE_MALICIOUS
1562 /**
1563  * Turn RPS service to act malicious.
1564  *
1565  * @param cls Closure
1566  * @param client The client that sent the message
1567  * @param msg The message header
1568  */
1569 static void
1570 handle_client_act_malicious (void *cls,
1571                              const struct GNUNET_MessageHeader *msg)
1572 {
1573   struct ClientContext *cli_ctx = cls;
1574   struct GNUNET_RPS_CS_ActMaliciousMessage *in_msg;
1575   struct GNUNET_PeerIdentity *peers;
1576   uint32_t num_mal_peers_sent;
1577   uint32_t num_mal_peers_old;
1578
1579   /* Check for protocol violation */
1580   if (sizeof (struct GNUNET_RPS_CS_ActMaliciousMessage) > ntohs (msg->size))
1581   {
1582     GNUNET_SERVICE_client_continue (cli_ctx->client);
1583     GNUNET_break_op (0);
1584   }
1585
1586   in_msg = (struct GNUNET_RPS_CS_ActMaliciousMessage *) msg;
1587   if ((ntohs (msg->size) - sizeof (struct GNUNET_RPS_CS_ActMaliciousMessage)) /
1588       sizeof (struct GNUNET_PeerIdentity) != ntohl (in_msg->num_peers))
1589   {
1590     LOG (GNUNET_ERROR_TYPE_ERROR,
1591         "message says it sends %" PRIu32 " peers, have space for %lu peers\n",
1592         ntohl (in_msg->num_peers),
1593         (ntohs (msg->size) - sizeof (struct GNUNET_RPS_CS_ActMaliciousMessage)) /
1594             sizeof (struct GNUNET_PeerIdentity));
1595     GNUNET_SERVICE_client_continue (cli_ctx->client);
1596     GNUNET_break_op (0);
1597   }
1598
1599
1600   /* Do actual logic */
1601   peers = (struct GNUNET_PeerIdentity *) &msg[1];
1602   mal_type = ntohl (in_msg->type);
1603   if (NULL == mal_peer_set)
1604     mal_peer_set = GNUNET_CONTAINER_multipeermap_create (1, GNUNET_NO);
1605
1606   LOG (GNUNET_ERROR_TYPE_DEBUG,
1607        "Now acting malicious type %" PRIu32 ", got %" PRIu32 " peers.\n",
1608        mal_type,
1609        ntohl (in_msg->num_peers));
1610
1611   if (1 == mal_type)
1612   { /* Try to maximise representation */
1613     /* Add other malicious peers to those we already know */
1614
1615     num_mal_peers_sent = ntohl (in_msg->num_peers);
1616     num_mal_peers_old = num_mal_peers;
1617     GNUNET_array_grow (mal_peers,
1618                        num_mal_peers,
1619                        num_mal_peers + num_mal_peers_sent);
1620     GNUNET_memcpy (&mal_peers[num_mal_peers_old],
1621             peers,
1622             num_mal_peers_sent * sizeof (struct GNUNET_PeerIdentity));
1623
1624     /* Add all mal peers to mal_peer_set */
1625     add_peer_array_to_set (&mal_peers[num_mal_peers_old],
1626                            num_mal_peers_sent,
1627                            mal_peer_set);
1628
1629     /* Substitute do_round () with do_mal_round () */
1630     GNUNET_SCHEDULER_cancel (do_round_task);
1631     do_round_task = GNUNET_SCHEDULER_add_now (&do_mal_round, NULL);
1632   }
1633
1634   else if ( (2 == mal_type) ||
1635             (3 == mal_type) )
1636   { /* Try to partition the network */
1637     /* Add other malicious peers to those we already know */
1638
1639     num_mal_peers_sent = ntohl (in_msg->num_peers) - 1;
1640     num_mal_peers_old = num_mal_peers;
1641     GNUNET_array_grow (mal_peers,
1642                        num_mal_peers,
1643                        num_mal_peers + num_mal_peers_sent);
1644     if (NULL != mal_peers &&
1645         0 != num_mal_peers)
1646     {
1647       GNUNET_memcpy (&mal_peers[num_mal_peers_old],
1648               peers,
1649               num_mal_peers_sent * sizeof (struct GNUNET_PeerIdentity));
1650
1651       /* Add all mal peers to mal_peer_set */
1652       add_peer_array_to_set (&mal_peers[num_mal_peers_old],
1653                              num_mal_peers_sent,
1654                              mal_peer_set);
1655     }
1656
1657     /* Store the one attacked peer */
1658     GNUNET_memcpy (&attacked_peer,
1659             &in_msg->attacked_peer,
1660             sizeof (struct GNUNET_PeerIdentity));
1661     /* Set the flag of the attacked peer to valid to avoid problems */
1662     if (GNUNET_NO == Peers_check_peer_known (&attacked_peer))
1663     {
1664       (void) Peers_issue_peer_liveliness_check (&attacked_peer);
1665     }
1666
1667     LOG (GNUNET_ERROR_TYPE_DEBUG,
1668          "Attacked peer is %s\n",
1669          GNUNET_i2s (&attacked_peer));
1670
1671     /* Substitute do_round () with do_mal_round () */
1672     GNUNET_SCHEDULER_cancel (do_round_task);
1673     do_round_task = GNUNET_SCHEDULER_add_now (&do_mal_round, NULL);
1674   }
1675   else if (0 == mal_type)
1676   { /* Stop acting malicious */
1677     GNUNET_array_grow (mal_peers, num_mal_peers, 0);
1678
1679     /* Substitute do_mal_round () with do_round () */
1680     GNUNET_SCHEDULER_cancel (do_round_task);
1681     do_round_task = GNUNET_SCHEDULER_add_now (&do_round, NULL);
1682   }
1683   else
1684   {
1685     GNUNET_break (0);
1686     GNUNET_SERVICE_client_continue (cli_ctx->client);
1687   }
1688   GNUNET_SERVICE_client_continue (cli_ctx->client);
1689 }
1690
1691
1692 /**
1693  * Send out PUSHes and PULLs maliciously.
1694  *
1695  * This is executed regylary.
1696  */
1697 static void
1698 do_mal_round (void *cls)
1699 {
1700   uint32_t num_pushes;
1701   uint32_t i;
1702   struct GNUNET_TIME_Relative time_next_round;
1703   struct AttackedPeer *tmp_att_peer;
1704
1705   LOG (GNUNET_ERROR_TYPE_DEBUG,
1706        "Going to execute next round maliciously type %" PRIu32 ".\n",
1707       mal_type);
1708   do_round_task = NULL;
1709   GNUNET_assert (mal_type <= 3);
1710   /* Do malicious actions */
1711   if (1 == mal_type)
1712   { /* Try to maximise representation */
1713
1714     /* The maximum of pushes we're going to send this round */
1715     num_pushes = GNUNET_MIN (GNUNET_MIN (push_limit,
1716                                          num_attacked_peers),
1717                              GNUNET_CONSTANTS_MAX_CADET_MESSAGE_SIZE);
1718
1719     LOG (GNUNET_ERROR_TYPE_DEBUG,
1720          "Going to send %" PRIu32 " pushes\n",
1721          num_pushes);
1722
1723     /* Send PUSHes to attacked peers */
1724     for (i = 0 ; i < num_pushes ; i++)
1725     {
1726       if (att_peers_tail == att_peer_index)
1727         att_peer_index = att_peers_head;
1728       else
1729         att_peer_index = att_peer_index->next;
1730
1731       send_push (&att_peer_index->peer_id);
1732     }
1733
1734     /* Send PULLs to some peers to learn about additional peers to attack */
1735     tmp_att_peer = att_peer_index;
1736     for (i = 0 ; i < num_pushes * alpha ; i++)
1737     {
1738       if (att_peers_tail == tmp_att_peer)
1739         tmp_att_peer = att_peers_head;
1740       else
1741         att_peer_index = tmp_att_peer->next;
1742
1743       send_pull_request (&tmp_att_peer->peer_id);
1744     }
1745   }
1746
1747
1748   else if (2 == mal_type)
1749   { /**
1750      * Try to partition the network
1751      * Send as many pushes to the attacked peer as possible
1752      * That is one push per round as it will ignore more.
1753      */
1754     (void) Peers_issue_peer_liveliness_check (&attacked_peer);
1755     if (GNUNET_YES == Peers_check_peer_flag (&attacked_peer, Peers_ONLINE))
1756       send_push (&attacked_peer);
1757   }
1758
1759
1760   if (3 == mal_type)
1761   { /* Combined attack */
1762
1763     /* Send PUSH to attacked peers */
1764     if (GNUNET_YES == Peers_check_peer_known (&attacked_peer))
1765     {
1766       (void) Peers_issue_peer_liveliness_check (&attacked_peer);
1767       if (GNUNET_YES == Peers_check_peer_flag (&attacked_peer, Peers_ONLINE))
1768       {
1769         LOG (GNUNET_ERROR_TYPE_DEBUG,
1770             "Goding to send push to attacked peer (%s)\n",
1771             GNUNET_i2s (&attacked_peer));
1772         send_push (&attacked_peer);
1773       }
1774     }
1775     (void) Peers_issue_peer_liveliness_check (&attacked_peer);
1776
1777     /* The maximum of pushes we're going to send this round */
1778     num_pushes = GNUNET_MIN (GNUNET_MIN (push_limit - 1,
1779                                          num_attacked_peers),
1780                              GNUNET_CONSTANTS_MAX_CADET_MESSAGE_SIZE);
1781
1782     LOG (GNUNET_ERROR_TYPE_DEBUG,
1783          "Going to send %" PRIu32 " pushes\n",
1784          num_pushes);
1785
1786     for (i = 0; i < num_pushes; i++)
1787     {
1788       if (att_peers_tail == att_peer_index)
1789         att_peer_index = att_peers_head;
1790       else
1791         att_peer_index = att_peer_index->next;
1792
1793       send_push (&att_peer_index->peer_id);
1794     }
1795
1796     /* Send PULLs to some peers to learn about additional peers to attack */
1797     tmp_att_peer = att_peer_index;
1798     for (i = 0; i < num_pushes * alpha; i++)
1799     {
1800       if (att_peers_tail == tmp_att_peer)
1801         tmp_att_peer = att_peers_head;
1802       else
1803         att_peer_index = tmp_att_peer->next;
1804
1805       send_pull_request (&tmp_att_peer->peer_id);
1806     }
1807   }
1808
1809   /* Schedule next round */
1810   time_next_round = compute_rand_delay (round_interval, 2);
1811
1812   //do_round_task = GNUNET_SCHEDULER_add_delayed (round_interval, &do_mal_round,
1813   //NULL);
1814   GNUNET_assert (NULL == do_round_task);
1815   do_round_task = GNUNET_SCHEDULER_add_delayed (time_next_round,
1816                                                 &do_mal_round, NULL);
1817   LOG (GNUNET_ERROR_TYPE_DEBUG, "Finished round\n");
1818 }
1819 #endif /* ENABLE_MALICIOUS */
1820
1821
1822 /**
1823  * Send out PUSHes and PULLs, possibly update #view, samplers.
1824  *
1825  * This is executed regylary.
1826  */
1827 static void
1828 do_round (void *cls)
1829 {
1830   uint32_t i;
1831   const struct GNUNET_PeerIdentity *view_array;
1832   unsigned int *permut;
1833   unsigned int a_peers; /* Number of peers we send pushes to */
1834   unsigned int b_peers; /* Number of peers we send pull requests to */
1835   uint32_t first_border;
1836   uint32_t second_border;
1837   struct GNUNET_PeerIdentity peer;
1838   struct GNUNET_PeerIdentity *update_peer;
1839
1840   LOG (GNUNET_ERROR_TYPE_DEBUG,
1841        "Going to execute next round.\n");
1842   do_round_task = NULL;
1843   LOG (GNUNET_ERROR_TYPE_DEBUG,
1844        "Printing view:\n");
1845   to_file (file_name_view_log,
1846            "___ new round ___");
1847   view_array = View_get_as_array ();
1848   for (i = 0; i < View_size (); i++)
1849   {
1850     LOG (GNUNET_ERROR_TYPE_DEBUG,
1851          "\t%s\n", GNUNET_i2s (&view_array[i]));
1852     to_file (file_name_view_log,
1853              "=%s\t(do round)",
1854              GNUNET_i2s_full (&view_array[i]));
1855   }
1856
1857
1858   /* Send pushes and pull requests */
1859   if (0 < View_size ())
1860   {
1861     permut = GNUNET_CRYPTO_random_permute (GNUNET_CRYPTO_QUALITY_STRONG,
1862                                            View_size ());
1863
1864     /* Send PUSHes */
1865     a_peers = ceil (alpha * View_size ());
1866
1867     LOG (GNUNET_ERROR_TYPE_DEBUG,
1868          "Going to send pushes to %u (ceil (%f * %u)) peers.\n",
1869          a_peers, alpha, View_size ());
1870     for (i = 0; i < a_peers; i++)
1871     {
1872       peer = view_array[permut[i]];
1873       if (0 != GNUNET_CRYPTO_cmp_peer_identity (&own_identity, &peer)) // TODO
1874       { // FIXME if this fails schedule/loop this for later
1875         send_push (&peer);
1876       }
1877     }
1878
1879     /* Send PULL requests */
1880     b_peers = ceil (beta * View_size ());
1881     first_border = a_peers;
1882     second_border = a_peers + b_peers;
1883     if (second_border > View_size ())
1884     {
1885       first_border = View_size () - b_peers;
1886       second_border = View_size ();
1887     }
1888     LOG (GNUNET_ERROR_TYPE_DEBUG,
1889         "Going to send pulls to %u (ceil (%f * %u)) peers.\n",
1890         b_peers, beta, View_size ());
1891     for (i = first_border; i < second_border; i++)
1892     {
1893       peer = view_array[permut[i]];
1894       if (0 != GNUNET_CRYPTO_cmp_peer_identity (&own_identity, &peer) &&
1895           GNUNET_NO == Peers_check_peer_flag (&peer, Peers_PULL_REPLY_PENDING)) // TODO
1896       { // FIXME if this fails schedule/loop this for later
1897         send_pull_request (&peer);
1898       }
1899     }
1900
1901     GNUNET_free (permut);
1902     permut = NULL;
1903   }
1904
1905
1906   /* Update view */
1907   /* TODO see how many peers are in push-/pull- list! */
1908
1909   if ((CustomPeerMap_size (push_map) <= alpha * View_size ()) &&
1910       (0 < CustomPeerMap_size (push_map)) &&
1911       (0 < CustomPeerMap_size (pull_map)))
1912   { /* If conditions for update are fulfilled, update */
1913     LOG (GNUNET_ERROR_TYPE_DEBUG, "Update of the view.\n");
1914
1915     uint32_t final_size;
1916     uint32_t peers_to_clean_size;
1917     struct GNUNET_PeerIdentity *peers_to_clean;
1918
1919     peers_to_clean = NULL;
1920     peers_to_clean_size = 0;
1921     GNUNET_array_grow (peers_to_clean, peers_to_clean_size, View_size ());
1922     GNUNET_memcpy (peers_to_clean,
1923             view_array,
1924             View_size () * sizeof (struct GNUNET_PeerIdentity));
1925
1926     /* Seems like recreating is the easiest way of emptying the peermap */
1927     View_clear ();
1928     to_file (file_name_view_log,
1929              "--- emptied ---");
1930
1931     first_border  = GNUNET_MIN (ceil (alpha * sampler_size_est_need),
1932                                 CustomPeerMap_size (push_map));
1933     second_border = first_border +
1934                     GNUNET_MIN (floor (beta  * sampler_size_est_need),
1935                                 CustomPeerMap_size (pull_map));
1936     final_size    = second_border +
1937       ceil ((1 - (alpha + beta)) * sampler_size_est_need);
1938
1939     /* Update view with peers received through PUSHes */
1940     permut = GNUNET_CRYPTO_random_permute (GNUNET_CRYPTO_QUALITY_STRONG,
1941                                            CustomPeerMap_size (push_map));
1942     for (i = 0; i < first_border; i++)
1943     {
1944       (void) insert_in_view (CustomPeerMap_get_peer_by_index (push_map,
1945                                                               permut[i]));
1946       to_file (file_name_view_log,
1947                "+%s\t(push list)",
1948                GNUNET_i2s_full (&view_array[i]));
1949       // TODO change the peer_flags accordingly
1950     }
1951     GNUNET_free (permut);
1952     permut = NULL;
1953
1954     /* Update view with peers received through PULLs */
1955     permut = GNUNET_CRYPTO_random_permute (GNUNET_CRYPTO_QUALITY_STRONG,
1956                                            CustomPeerMap_size (pull_map));
1957     for (i = first_border; i < second_border; i++)
1958     {
1959       (void) insert_in_view (CustomPeerMap_get_peer_by_index (pull_map,
1960             permut[i - first_border]));
1961       to_file (file_name_view_log,
1962                "+%s\t(pull list)",
1963                GNUNET_i2s_full (&view_array[i]));
1964       // TODO change the peer_flags accordingly
1965     }
1966     GNUNET_free (permut);
1967     permut = NULL;
1968
1969     /* Update view with peers from history */
1970     RPS_sampler_get_n_rand_peers (prot_sampler,
1971                                   hist_update,
1972                                   NULL,
1973                                   final_size - second_border);
1974     // TODO change the peer_flags accordingly
1975
1976     for (i = 0; i < View_size (); i++)
1977       rem_from_list (&peers_to_clean, &peers_to_clean_size, &view_array[i]);
1978
1979     /* Clean peers that were removed from the view */
1980     for (i = 0; i < peers_to_clean_size; i++)
1981     {
1982       to_file (file_name_view_log,
1983                "-%s",
1984                GNUNET_i2s_full (&peers_to_clean[i]));
1985       clean_peer (&peers_to_clean[i]);
1986       //peer_destroy_channel_send (sender);
1987     }
1988
1989     GNUNET_array_grow (peers_to_clean, peers_to_clean_size, 0);
1990   }
1991   else
1992   {
1993     LOG (GNUNET_ERROR_TYPE_DEBUG, "No update of the view.\n");
1994   }
1995   // TODO independent of that also get some peers from CADET_get_peers()?
1996
1997   LOG (GNUNET_ERROR_TYPE_DEBUG,
1998        "Received %u pushes and %u pulls last round (alpha (%.2f) * view_size (%u) = %.2f)\n",
1999        CustomPeerMap_size (push_map),
2000        CustomPeerMap_size (pull_map),
2001        alpha,
2002        View_size (),
2003        alpha * View_size ());
2004
2005   /* Update samplers */
2006   for (i = 0; i < CustomPeerMap_size (push_map); i++)
2007   {
2008     update_peer = CustomPeerMap_get_peer_by_index (push_map, i);
2009     LOG (GNUNET_ERROR_TYPE_DEBUG,
2010          "Updating with peer %s from push list\n",
2011          GNUNET_i2s (update_peer));
2012     insert_in_sampler (NULL, update_peer);
2013     clean_peer (update_peer); /* This cleans only if it is not in the view */
2014     //peer_destroy_channel_send (sender);
2015   }
2016
2017   for (i = 0; i < CustomPeerMap_size (pull_map); i++)
2018   {
2019     LOG (GNUNET_ERROR_TYPE_DEBUG,
2020          "Updating with peer %s from pull list\n",
2021          GNUNET_i2s (CustomPeerMap_get_peer_by_index (pull_map, i)));
2022     insert_in_sampler (NULL, CustomPeerMap_get_peer_by_index (pull_map, i));
2023     /* This cleans only if it is not in the view */
2024     clean_peer (CustomPeerMap_get_peer_by_index (pull_map, i));
2025     //peer_destroy_channel_send (sender);
2026   }
2027
2028
2029   /* Empty push/pull lists */
2030   CustomPeerMap_clear (push_map);
2031   CustomPeerMap_clear (pull_map);
2032
2033   struct GNUNET_TIME_Relative time_next_round;
2034
2035   time_next_round = compute_rand_delay (round_interval, 2);
2036
2037   /* Schedule next round */
2038   do_round_task = GNUNET_SCHEDULER_add_delayed (time_next_round,
2039                                                 &do_round, NULL);
2040   LOG (GNUNET_ERROR_TYPE_DEBUG, "Finished round\n");
2041 }
2042
2043
2044 /**
2045  * This is called from GNUNET_CADET_get_peers().
2046  *
2047  * It is called on every peer(ID) that cadet somehow has contact with.
2048  * We use those to initialise the sampler.
2049  */
2050 void
2051 init_peer_cb (void *cls,
2052               const struct GNUNET_PeerIdentity *peer,
2053               int tunnel, // "Do we have a tunnel towards this peer?"
2054               unsigned int n_paths, // "Number of known paths towards this peer"
2055               unsigned int best_path) // "How long is the best path?
2056                                       // (0 = unknown, 1 = ourselves, 2 = neighbor)"
2057 {
2058   if (NULL != peer)
2059   {
2060     LOG (GNUNET_ERROR_TYPE_DEBUG,
2061          "Got peer_id %s from cadet\n",
2062          GNUNET_i2s (peer));
2063     got_peer (peer);
2064   }
2065 }
2066
2067 /**
2068  * @brief Iterator function over stored, valid peers.
2069  *
2070  * We initialise the sampler with those.
2071  *
2072  * @param cls the closure
2073  * @param peer the peer id
2074  * @return #GNUNET_YES if we should continue to
2075  *         iterate,
2076  *         #GNUNET_NO if not.
2077  */
2078 static int
2079 valid_peers_iterator (void *cls,
2080                       const struct GNUNET_PeerIdentity *peer)
2081 {
2082   if (NULL != peer)
2083   {
2084     LOG (GNUNET_ERROR_TYPE_DEBUG,
2085          "Got stored, valid peer %s\n",
2086          GNUNET_i2s (peer));
2087     got_peer (peer);
2088   }
2089   return GNUNET_YES;
2090 }
2091
2092
2093 /**
2094  * Iterator over peers from peerinfo.
2095  *
2096  * @param cls closure
2097  * @param peer id of the peer, NULL for last call
2098  * @param hello hello message for the peer (can be NULL)
2099  * @param error message
2100  */
2101 void
2102 process_peerinfo_peers (void *cls,
2103                         const struct GNUNET_PeerIdentity *peer,
2104                         const struct GNUNET_HELLO_Message *hello,
2105                         const char *err_msg)
2106 {
2107   if (NULL != peer)
2108   {
2109     LOG (GNUNET_ERROR_TYPE_DEBUG,
2110          "Got peer_id %s from peerinfo\n",
2111          GNUNET_i2s (peer));
2112     got_peer (peer);
2113   }
2114 }
2115
2116
2117 /**
2118  * Task run during shutdown.
2119  *
2120  * @param cls unused
2121  */
2122 static void
2123 shutdown_task (void *cls)
2124 {
2125   struct ClientContext *client_ctx;
2126   struct ReplyCls *reply_cls;
2127
2128   LOG (GNUNET_ERROR_TYPE_DEBUG,
2129        "RPS is going down\n");
2130
2131   /* Clean all clients */
2132   for (client_ctx = cli_ctx_head;
2133        NULL != cli_ctx_head;
2134        client_ctx = cli_ctx_head)
2135   {
2136     /* Clean pending requests to the sampler */
2137     for (reply_cls = client_ctx->rep_cls_head;
2138          NULL != client_ctx->rep_cls_head;
2139          reply_cls = client_ctx->rep_cls_head)
2140     {
2141       RPS_sampler_request_cancel (reply_cls->req_handle);
2142       GNUNET_CONTAINER_DLL_remove (client_ctx->rep_cls_head,
2143                                    client_ctx->rep_cls_tail,
2144                                    reply_cls);
2145       GNUNET_free (reply_cls);
2146     }
2147     GNUNET_MQ_destroy (client_ctx->mq);
2148     GNUNET_CONTAINER_DLL_remove (cli_ctx_head, cli_ctx_tail, client_ctx);
2149     GNUNET_free (client_ctx);
2150   }
2151   GNUNET_PEERINFO_notify_cancel (peerinfo_notify_handle);
2152   GNUNET_PEERINFO_disconnect (peerinfo_handle);
2153
2154   if (NULL != do_round_task)
2155   {
2156     GNUNET_SCHEDULER_cancel (do_round_task);
2157     do_round_task = NULL;
2158   }
2159
2160   Peers_terminate ();
2161
2162   GNUNET_NSE_disconnect (nse);
2163   RPS_sampler_destroy (prot_sampler);
2164   RPS_sampler_destroy (client_sampler);
2165   GNUNET_CADET_disconnect (cadet_handle);
2166   View_destroy ();
2167   CustomPeerMap_destroy (push_map);
2168   CustomPeerMap_destroy (pull_map);
2169   #ifdef ENABLE_MALICIOUS
2170   struct AttackedPeer *tmp_att_peer;
2171   GNUNET_free (file_name_view_log);
2172   GNUNET_array_grow (mal_peers, num_mal_peers, 0);
2173   if (NULL != mal_peer_set)
2174     GNUNET_CONTAINER_multipeermap_destroy (mal_peer_set);
2175   if (NULL != att_peer_set)
2176     GNUNET_CONTAINER_multipeermap_destroy (att_peer_set);
2177   while (NULL != att_peers_head)
2178   {
2179     tmp_att_peer = att_peers_head;
2180     GNUNET_CONTAINER_DLL_remove (att_peers_head, att_peers_tail, tmp_att_peer);
2181   }
2182   #endif /* ENABLE_MALICIOUS */
2183 }
2184
2185
2186 /**
2187  * Handle client connecting to the service.
2188  *
2189  * @param cls NULL
2190  * @param client the new client
2191  * @param mq the message queue of @a client
2192  * @return @a client
2193  */
2194 static void *
2195 client_connect_cb (void *cls,
2196                    struct GNUNET_SERVICE_Client *client,
2197                    struct GNUNET_MQ_Handle *mq)
2198 {
2199   struct ClientContext *cli_ctx;
2200
2201   LOG (GNUNET_ERROR_TYPE_DEBUG,
2202        "Client connected\n");
2203   if (NULL == client)
2204     return client; /* Server was destroyed before a client connected. Shutting down */
2205   cli_ctx = GNUNET_new (struct ClientContext);
2206   cli_ctx->mq = GNUNET_SERVICE_client_get_mq (client);
2207   cli_ctx->client = client;
2208   GNUNET_CONTAINER_DLL_insert (cli_ctx_head,
2209                                cli_ctx_tail,
2210                                cli_ctx);
2211   return cli_ctx;
2212 }
2213
2214 /**
2215  * Callback called when a client disconnected from the service
2216  *
2217  * @param cls closure for the service
2218  * @param c the client that disconnected
2219  * @param internal_cls should be equal to @a c
2220  */
2221 static void
2222 client_disconnect_cb (void *cls,
2223                                   struct GNUNET_SERVICE_Client *client,
2224                                   void *internal_cls)
2225 {
2226   struct ClientContext *cli_ctx = internal_cls;
2227
2228   GNUNET_assert (client == cli_ctx->client);
2229   if (NULL == client)
2230   {/* shutdown task */
2231     while (NULL != cli_ctx_head)
2232       destroy_cli_ctx (cli_ctx_head);
2233   }
2234   else
2235   {
2236     destroy_cli_ctx (cli_ctx);
2237   }
2238 }
2239
2240
2241 /**
2242  * Handle random peer sampling clients.
2243  *
2244  * @param cls closure
2245  * @param c configuration to use
2246  * @param service the initialized service
2247  */
2248 static void
2249 run (void *cls,
2250      const struct GNUNET_CONFIGURATION_Handle *c,
2251      struct GNUNET_SERVICE_Handle *service)
2252 {
2253   static const struct GNUNET_CADET_MessageHandler cadet_handlers[] = {
2254     {&handle_peer_check       , GNUNET_MESSAGE_TYPE_RPS_PP_CHECK_LIVE,
2255       sizeof (struct GNUNET_MessageHeader)},
2256     {&handle_peer_push        , GNUNET_MESSAGE_TYPE_RPS_PP_PUSH,
2257       sizeof (struct GNUNET_MessageHeader)},
2258     {&handle_peer_pull_request, GNUNET_MESSAGE_TYPE_RPS_PP_PULL_REQUEST,
2259       sizeof (struct GNUNET_MessageHeader)},
2260     {&handle_peer_pull_reply  , GNUNET_MESSAGE_TYPE_RPS_PP_PULL_REPLY, 0},
2261     {NULL, 0, 0}
2262   };
2263
2264   int size;
2265   int out_size;
2266   char* fn_valid_peers;
2267   struct GNUNET_HashCode port;
2268
2269   GNUNET_log_setup ("rps", GNUNET_error_type_to_string (GNUNET_ERROR_TYPE_DEBUG), NULL);
2270   cfg = c;
2271
2272
2273   /* Get own ID */
2274   GNUNET_CRYPTO_get_peer_identity (cfg, &own_identity); // TODO check return value
2275   GNUNET_log (GNUNET_ERROR_TYPE_INFO,
2276               "STARTING SERVICE (rps) for peer [%s]\n",
2277               GNUNET_i2s (&own_identity));
2278   #ifdef ENABLE_MALICIOUS
2279   GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
2280               "Malicious execution compiled in.\n");
2281   #endif /* ENABLE_MALICIOUS */
2282
2283
2284
2285   /* Get time interval from the configuration */
2286   if (GNUNET_OK != GNUNET_CONFIGURATION_get_value_time (cfg, "RPS",
2287                                                         "ROUNDINTERVAL",
2288                                                         &round_interval))
2289   {
2290     GNUNET_log_config_missing (GNUNET_ERROR_TYPE_ERROR,
2291                                "RPS", "ROUNDINTERVAL");
2292     GNUNET_SCHEDULER_shutdown ();
2293     return;
2294   }
2295
2296   /* Get initial size of sampler/view from the configuration */
2297   if (GNUNET_OK !=
2298       GNUNET_CONFIGURATION_get_value_number (cfg, "RPS", "INITSIZE",
2299         (long long unsigned int *) &sampler_size_est_need))
2300   {
2301     GNUNET_log_config_missing (GNUNET_ERROR_TYPE_ERROR,
2302                                "RPS", "INITSIZE");
2303     GNUNET_SCHEDULER_shutdown ();
2304     return;
2305   }
2306   LOG (GNUNET_ERROR_TYPE_DEBUG, "INITSIZE is %u\n", sampler_size_est_need);
2307
2308   if (GNUNET_OK !=
2309       GNUNET_CONFIGURATION_get_value_filename (cfg,
2310                                                "rps",
2311                                                "FILENAME_VALID_PEERS",
2312                                                &fn_valid_peers))
2313   {
2314     GNUNET_log_config_missing (GNUNET_ERROR_TYPE_ERROR,
2315                                "rps", "FILENAME_VALID_PEERS");
2316   }
2317
2318
2319   View_create (4);
2320
2321   /* file_name_view_log */
2322   if (GNUNET_OK != GNUNET_DISK_directory_create ("/tmp/rps/"))
2323   {
2324     LOG (GNUNET_ERROR_TYPE_WARNING,
2325          "Failed to create directory /tmp/rps/\n");
2326   }
2327
2328   size = (14 + strlen (GNUNET_i2s_full (&own_identity)) + 1) * sizeof (char);
2329   file_name_view_log = GNUNET_malloc (size);
2330   out_size = GNUNET_snprintf (file_name_view_log,
2331                               size,
2332                               "/tmp/rps/view-%s",
2333                               GNUNET_i2s_full (&own_identity));
2334   if (size < out_size ||
2335       0 > out_size)
2336   {
2337     LOG (GNUNET_ERROR_TYPE_WARNING,
2338          "Failed to write string to buffer (size: %i, out_size: %i)\n",
2339          size,
2340          out_size);
2341   }
2342
2343
2344   /* connect to NSE */
2345   nse = GNUNET_NSE_connect (cfg, nse_callback, NULL);
2346
2347
2348   alpha = 0.45;
2349   beta  = 0.45;
2350
2351
2352   /* Initialise cadet */
2353   cadet_handle = GNUNET_CADET_connect (cfg,
2354                                        cls,
2355                                        &cleanup_destroyed_channel,
2356                                        cadet_handlers);
2357   GNUNET_assert (NULL != cadet_handle);
2358   GNUNET_CRYPTO_hash (GNUNET_APPLICATION_PORT_RPS,
2359                       strlen (GNUNET_APPLICATION_PORT_RPS),
2360                       &port);
2361   GNUNET_CADET_open_port (cadet_handle,
2362                           &port,
2363                           &Peers_handle_inbound_channel, cls);
2364
2365
2366   peerinfo_handle = GNUNET_PEERINFO_connect (cfg);
2367   Peers_initialise (fn_valid_peers, cadet_handle, &own_identity);
2368   GNUNET_free (fn_valid_peers);
2369
2370   /* Initialise sampler */
2371   struct GNUNET_TIME_Relative half_round_interval;
2372   struct GNUNET_TIME_Relative  max_round_interval;
2373
2374   half_round_interval = GNUNET_TIME_relative_multiply (round_interval, .5);
2375   max_round_interval = GNUNET_TIME_relative_add (round_interval, half_round_interval);
2376
2377   prot_sampler =   RPS_sampler_init     (sampler_size_est_need, max_round_interval);
2378   client_sampler = RPS_sampler_mod_init (sampler_size_est_need, max_round_interval);
2379
2380   /* Initialise push and pull maps */
2381   push_map = CustomPeerMap_create (4);
2382   pull_map = CustomPeerMap_create (4);
2383
2384
2385   //LOG (GNUNET_ERROR_TYPE_DEBUG, "Requesting peers from CADET\n");
2386   //GNUNET_CADET_get_peers (cadet_handle, &init_peer_cb, NULL);
2387   // TODO send push/pull to each of those peers?
2388   // TODO read stored valid peers from last run
2389   LOG (GNUNET_ERROR_TYPE_DEBUG, "Requesting stored valid peers\n");
2390   Peers_get_valid_peers (valid_peers_iterator, NULL);
2391
2392   peerinfo_notify_handle = GNUNET_PEERINFO_notify (cfg,
2393                                                    GNUNET_NO,
2394                                                    process_peerinfo_peers,
2395                                                    NULL);
2396
2397   LOG (GNUNET_ERROR_TYPE_INFO, "Ready to receive requests from clients\n");
2398
2399   do_round_task = GNUNET_SCHEDULER_add_now (&do_round, NULL);
2400   LOG (GNUNET_ERROR_TYPE_DEBUG, "Scheduled first round\n");
2401
2402   GNUNET_SCHEDULER_add_shutdown (&shutdown_task, NULL);
2403 }
2404
2405
2406 /**
2407  * Define "main" method using service macro.
2408  */
2409 GNUNET_SERVICE_MAIN
2410 ("rps",
2411  GNUNET_SERVICE_OPTION_NONE,
2412  &run,
2413  &client_connect_cb,
2414  &client_disconnect_cb,
2415  NULL,
2416  GNUNET_MQ_hd_fixed_size (client_request,
2417    GNUNET_MESSAGE_TYPE_RPS_CS_REQUEST,
2418    struct GNUNET_RPS_CS_RequestMessage,
2419    NULL),
2420  GNUNET_MQ_hd_fixed_size (client_request_cancel,
2421    GNUNET_MESSAGE_TYPE_RPS_CS_REQUEST_CANCEL,
2422    struct GNUNET_RPS_CS_RequestCancelMessage,
2423    NULL),
2424  GNUNET_MQ_hd_fixed_size (client_seed,
2425    GNUNET_MESSAGE_TYPE_RPS_CS_SEED,
2426    struct GNUNET_MessageHeader,
2427    NULL),
2428 #ifdef ENABLE_MALICIOUS
2429  GNUNET_MQ_hd_fixed_size (client_act_malicious,
2430    GNUNET_MESSAGE_TYPE_RPS_ACT_MALICIOUS,
2431    struct GNUNET_MessageHeader,
2432    NULL),
2433 #endif /* ENABLE_MALICIOUS */
2434  GNUNET_MQ_handler_end());
2435
2436 /* end of gnunet-service-rps.c */