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   GNUNET_CONTAINER_DLL_remove (cli_ctx->rep_cls_head,
 930                                cli_ctx->rep_cls_tail,
 931                                rep_cls);
 932   GNUNET_free (rep_cls);
 933 }
 934
 935
 936 static void
 937 destroy_cli_ctx (struct ClientContext *cli_ctx)
 938 {
 939   GNUNET_assert (NULL != cli_ctx);
 940   if (NULL != cli_ctx->rep_cls_head)
 941   {
 942     LOG (GNUNET_ERROR_TYPE_WARNING,
 943          "Trying to destroy the context of a client that still has pending requests. Going to clean those\n");
 944     while (NULL != cli_ctx->rep_cls_head)
 945       destroy_reply_cls (cli_ctx->rep_cls_head);
 946   }
 947   GNUNET_CONTAINER_DLL_remove (cli_ctx_head,
 948                                cli_ctx_tail,
 949                                cli_ctx);
 950   GNUNET_free (cli_ctx);
 951 }
 952
 953
 954 /**
 955  * Function called by NSE.
 956  *
 957  * Updates sizes of sampler list and view and adapt those lists
 958  * accordingly.
 959  */
 960 static void
 961 nse_callback (void *cls,
 962               struct GNUNET_TIME_Absolute timestamp,
 963               double logestimate, double std_dev)
 964 {
 965   double estimate;
 966   //double scale; // TODO this might go gloabal/config
 967
 968   LOG (GNUNET_ERROR_TYPE_DEBUG,
 969        "Received a ns estimate - logest: %f, std_dev: %f (old_size: %u)\n",
 970        logestimate, std_dev, RPS_sampler_get_size (prot_sampler));
 971   //scale = .01;
 972   estimate = GNUNET_NSE_log_estimate_to_n (logestimate);
 973   // GNUNET_NSE_log_estimate_to_n (logestimate);
 974   estimate = pow (estimate, 1.0 / 3);
 975   // TODO add if std_dev is a number
 976   // estimate += (std_dev * scale);
 977   if (2 < ceil (estimate))
 978   {
 979     LOG (GNUNET_ERROR_TYPE_DEBUG, "Changing estimate to %f\n", estimate);
 980     sampler_size_est_need = estimate;
 981   } else
 982     LOG (GNUNET_ERROR_TYPE_DEBUG, "Not using estimate %f\n", estimate);
 983
 984   /* If the NSE has changed adapt the lists accordingly */
 985   resize_wrapper (prot_sampler, sampler_size_est_need);
 986   client_resize_wrapper ();
 987 }
 988
 989
 990 /**
 991  * Callback called once the requested PeerIDs are ready.
 992  *
 993  * Sends those to the requesting client.
 994  */
 995 static void
 996 client_respond (void *cls,
 997                 struct GNUNET_PeerIdentity *peer_ids,
 998                 uint32_t num_peers)
 999 {
1000   struct ReplyCls *reply_cls = cls;
1001   uint32_t i;
1002   struct GNUNET_MQ_Envelope *ev;
1003   struct GNUNET_RPS_CS_ReplyMessage *out_msg;
1004   uint32_t size_needed;
1005   struct ClientContext *cli_ctx;
1006
1007   GNUNET_assert (NULL != reply_cls);
1008   LOG (GNUNET_ERROR_TYPE_DEBUG,
1009        "sampler returned %" PRIu32 " peers:\n",
1010        num_peers);
1011   for (i = 0; i < num_peers; i++)
1012   {
1013     LOG (GNUNET_ERROR_TYPE_DEBUG,
1014          "  %" PRIu32 ": %s\n",
1015          i,
1016          GNUNET_i2s (&peer_ids[i]));
1017   }
1018
1019   size_needed = sizeof (struct GNUNET_RPS_CS_ReplyMessage) +
1020                 num_peers * sizeof (struct GNUNET_PeerIdentity);
1021
1022   GNUNET_assert (GNUNET_SERVER_MAX_MESSAGE_SIZE >= size_needed);
1023
1024   ev = GNUNET_MQ_msg_extra (out_msg,
1025                             num_peers * sizeof (struct GNUNET_PeerIdentity),
1026                             GNUNET_MESSAGE_TYPE_RPS_CS_REPLY);
1027   out_msg->num_peers = htonl (num_peers);
1028   out_msg->id = htonl (reply_cls->id);
1029
1030   GNUNET_memcpy (&out_msg[1],
1031           peer_ids,
1032           num_peers * sizeof (struct GNUNET_PeerIdentity));
1033   GNUNET_free (peer_ids);
1034
1035   cli_ctx = reply_cls->cli_ctx;
1036   GNUNET_assert (NULL != cli_ctx);
1037   destroy_reply_cls (reply_cls);
1038   GNUNET_MQ_send (cli_ctx->mq, ev);
1039 }
1040
1041
1042 /**
1043  * Handle RPS request from the client.
1044  *
1045  * @param cls closure
1046  * @param client identification of the client
1047  * @param message the actual message
1048  */
1049 static void
1050 handle_client_request (void *cls,
1051                        const struct GNUNET_RPS_CS_RequestMessage *msg)
1052 {
1053   struct ClientContext *cli_ctx = cls;
1054   uint32_t num_peers;
1055   uint32_t size_needed;
1056   struct ReplyCls *reply_cls;
1057   uint32_t i;
1058
1059   num_peers = ntohl (msg->num_peers);
1060   size_needed = sizeof (struct GNUNET_RPS_CS_RequestMessage) +
1061                 num_peers * sizeof (struct GNUNET_PeerIdentity);
1062
1063   if (GNUNET_SERVER_MAX_MESSAGE_SIZE < size_needed)
1064   {
1065     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
1066                 "Message received from client has size larger than expected\n");
1067     GNUNET_SERVICE_client_drop (cli_ctx->client);
1068     return;
1069   }
1070
1071   for (i = 0 ; i < num_peers ; i++)
1072     est_request_rate();
1073
1074   LOG (GNUNET_ERROR_TYPE_DEBUG,
1075        "Client requested %" PRIu32 " random peer(s).\n",
1076        num_peers);
1077
1078   reply_cls = GNUNET_new (struct ReplyCls);
1079   reply_cls->id = ntohl (msg->id);
1080   reply_cls->cli_ctx = cli_ctx;
1081   reply_cls->req_handle = RPS_sampler_get_n_rand_peers (client_sampler,
1082                                                         client_respond,
1083                                                         reply_cls,
1084                                                         num_peers);
1085
1086   GNUNET_assert (NULL != cli_ctx);
1087   GNUNET_CONTAINER_DLL_insert (cli_ctx->rep_cls_head,
1088                                cli_ctx->rep_cls_tail,
1089                                reply_cls);
1090   GNUNET_SERVICE_client_continue (cli_ctx->client);
1091 }
1092
1093
1094 /**
1095  * @brief Handle a message that requests the cancellation of a request
1096  *
1097  * @param cls unused
1098  * @param client the client that requests the cancellation
1099  * @param message the message containing the id of the request
1100  */
1101 static void
1102 handle_client_request_cancel (void *cls,
1103                           const struct GNUNET_RPS_CS_RequestCancelMessage *msg)
1104 {
1105   struct ClientContext *cli_ctx = cls;
1106   struct ReplyCls *rep_cls;
1107
1108   GNUNET_assert (NULL != cli_ctx);
1109   GNUNET_assert (NULL != cli_ctx->rep_cls_head);
1110   rep_cls = cli_ctx->rep_cls_head;
1111   LOG (GNUNET_ERROR_TYPE_DEBUG,
1112       "Client cancels request with id %" PRIu32 "\n",
1113       ntohl (msg->id));
1114   while ( (NULL != rep_cls->next) &&
1115           (rep_cls->id != ntohl (msg->id)) )
1116     rep_cls = rep_cls->next;
1117   GNUNET_assert (rep_cls->id == ntohl (msg->id));
1118   RPS_sampler_request_cancel (rep_cls->req_handle);
1119   destroy_reply_cls (rep_cls);
1120   GNUNET_SERVICE_client_continue (cli_ctx->client);
1121 }
1122
1123
1124 /**
1125  * Handle seed from the client.
1126  *
1127  * @param cls closure
1128  * @param client identification of the client
1129  * @param message the actual message
1130  */
1131 static void
1132 handle_client_seed (void *cls,
1133                     const struct GNUNET_MessageHeader *message)
1134 {
1135   struct ClientContext *cli_ctx = cls;
1136   struct GNUNET_RPS_CS_SeedMessage *in_msg;
1137   struct GNUNET_PeerIdentity *peers;
1138   uint32_t num_peers;
1139   uint32_t i;
1140
1141   if (sizeof (struct GNUNET_RPS_CS_SeedMessage) > ntohs (message->size))
1142   {
1143     GNUNET_SERVICE_client_drop (cli_ctx->client);
1144     GNUNET_break_op (0);
1145     return;
1146   }
1147
1148   in_msg = (struct GNUNET_RPS_CS_SeedMessage *) message;
1149   num_peers = ntohl (in_msg->num_peers);
1150   peers = (struct GNUNET_PeerIdentity *) &in_msg[1];
1151   //peers = GNUNET_new_array (num_peers, struct GNUNET_PeerIdentity);
1152   //GNUNET_memcpy (peers, &in_msg[1], num_peers * sizeof (struct GNUNET_PeerIdentity));
1153
1154   if ((ntohs (message->size) - sizeof (struct GNUNET_RPS_CS_SeedMessage)) /
1155       sizeof (struct GNUNET_PeerIdentity) != num_peers)
1156   {
1157     GNUNET_break_op (0);
1158     GNUNET_SERVICE_client_drop (cli_ctx->client);
1159     return;
1160   }
1161
1162   LOG (GNUNET_ERROR_TYPE_DEBUG,
1163        "Client seeded peers:\n");
1164   print_peer_list (peers, num_peers);
1165
1166   for (i = 0; i < num_peers; i++)
1167   {
1168     LOG (GNUNET_ERROR_TYPE_DEBUG,
1169          "Updating samplers with seed %" PRIu32 ": %s\n",
1170          i,
1171          GNUNET_i2s (&peers[i]));
1172
1173     got_peer (&peers[i]);
1174
1175     //RPS_sampler_update (prot_sampler,   &peers[i]);
1176     //RPS_sampler_update (client_sampler, &peers[i]);
1177   }
1178
1179   ////GNUNET_free (peers);
1180
1181   GNUNET_SERVICE_client_drop (cli_ctx->client);
1182 }
1183
1184 /**
1185  * Handle a CHECK_LIVE message from another peer.
1186  *
1187  * This does nothing. But without calling #GNUNET_CADET_receive_done()
1188  * the channel is blocked for all other communication.
1189  *
1190  * @param cls Closure
1191  * @param channel The channel the CHECK was received over
1192  * @param channel_ctx The context associated with this channel
1193  * @param msg The message header
1194  */
1195 static int
1196 handle_peer_check (void *cls,
1197                   struct GNUNET_CADET_Channel *channel,
1198                   void **channel_ctx,
1199                   const struct GNUNET_MessageHeader *msg)
1200 {
1201   GNUNET_CADET_receive_done (channel);
1202   return GNUNET_OK;
1203 }
1204
1205 /**
1206  * Handle a PUSH message from another peer.
1207  *
1208  * Check the proof of work and store the PeerID
1209  * in the temporary list for pushed PeerIDs.
1210  *
1211  * @param cls Closure
1212  * @param channel The channel the PUSH was received over
1213  * @param channel_ctx The context associated with this channel
1214  * @param msg The message header
1215  */
1216 static int
1217 handle_peer_push (void *cls,
1218                   struct GNUNET_CADET_Channel *channel,
1219                   void **channel_ctx,
1220                   const struct GNUNET_MessageHeader *msg)
1221 {
1222   const struct GNUNET_PeerIdentity *peer;
1223
1224   // (check the proof of work (?))
1225
1226   peer = (const struct GNUNET_PeerIdentity *)
1227     GNUNET_CADET_channel_get_info (channel, GNUNET_CADET_OPTION_PEER);
1228   // FIXME wait for cadet to change this function
1229
1230   LOG (GNUNET_ERROR_TYPE_DEBUG,
1231        "Received PUSH (%s)\n",
1232        GNUNET_i2s (peer));
1233
1234   #ifdef ENABLE_MALICIOUS
1235   struct AttackedPeer *tmp_att_peer;
1236
1237   if ( (1 == mal_type) ||
1238        (3 == mal_type) )
1239   { /* Try to maximise representation */
1240     tmp_att_peer = GNUNET_new (struct AttackedPeer);
1241     GNUNET_memcpy (&tmp_att_peer->peer_id, peer, sizeof (struct GNUNET_PeerIdentity));
1242     if (NULL == att_peer_set)
1243       att_peer_set = GNUNET_CONTAINER_multipeermap_create (1, GNUNET_NO);
1244     if (GNUNET_NO == GNUNET_CONTAINER_multipeermap_contains (att_peer_set,
1245                                                              peer))
1246     {
1247       GNUNET_CONTAINER_DLL_insert (att_peers_head,
1248                                    att_peers_tail,
1249                                    tmp_att_peer);
1250       add_peer_array_to_set (peer, 1, att_peer_set);
1251     }
1252     GNUNET_CADET_receive_done (channel);
1253     return GNUNET_OK;
1254   }
1255
1256
1257   else if (2 == mal_type)
1258   { /* We attack one single well-known peer - simply ignore */
1259     GNUNET_CADET_receive_done (channel);
1260     return GNUNET_OK;
1261   }
1262   #endif /* ENABLE_MALICIOUS */
1263
1264   /* Add the sending peer to the push_map */
1265   CustomPeerMap_put (push_map, peer);
1266
1267   GNUNET_CADET_receive_done (channel);
1268   return GNUNET_OK;
1269 }
1270
1271
1272 /**
1273  * Handle PULL REQUEST request message from another peer.
1274  *
1275  * Reply with the view of PeerIDs.
1276  *
1277  * @param cls Closure
1278  * @param channel The channel the PULL REQUEST was received over
1279  * @param channel_ctx The context associated with this channel
1280  * @param msg The message header
1281  */
1282 static int
1283 handle_peer_pull_request (void *cls,
1284                           struct GNUNET_CADET_Channel *channel,
1285                           void **channel_ctx,
1286                           const struct GNUNET_MessageHeader *msg)
1287 {
1288   struct GNUNET_PeerIdentity *peer;
1289   const struct GNUNET_PeerIdentity *view_array;
1290
1291   peer = (struct GNUNET_PeerIdentity *)
1292     GNUNET_CADET_channel_get_info (channel,
1293                                    GNUNET_CADET_OPTION_PEER);
1294   // FIXME wait for cadet to change this function
1295
1296   LOG (GNUNET_ERROR_TYPE_DEBUG, "Received PULL REQUEST (%s)\n", GNUNET_i2s (peer));
1297
1298   #ifdef ENABLE_MALICIOUS
1299   if (1 == mal_type
1300       || 3 == mal_type)
1301   { /* Try to maximise representation */
1302     send_pull_reply (peer, mal_peers, num_mal_peers);
1303     GNUNET_CADET_receive_done (channel);
1304     return GNUNET_OK;
1305   }
1306
1307   else if (2 == mal_type)
1308   { /* Try to partition network */
1309     if (0 == GNUNET_CRYPTO_cmp_peer_identity (&attacked_peer, peer))
1310     {
1311       send_pull_reply (peer, mal_peers, num_mal_peers);
1312     }
1313     GNUNET_CADET_receive_done (channel);
1314     return GNUNET_OK;
1315   }
1316   #endif /* ENABLE_MALICIOUS */
1317
1318   view_array = View_get_as_array ();
1319
1320   send_pull_reply (peer, view_array, View_size ());
1321
1322   GNUNET_CADET_receive_done (channel);
1323   return GNUNET_OK;
1324 }
1325
1326
1327 /**
1328  * Handle PULL REPLY message from another peer.
1329  *
1330  * Check whether we sent a corresponding request and
1331  * whether this reply is the first one.
1332  *
1333  * @param cls Closure
1334  * @param channel The channel the PUSH was received over
1335  * @param channel_ctx The context associated with this channel
1336  * @param msg The message header
1337  */
1338 static int
1339 handle_peer_pull_reply (void *cls,
1340                         struct GNUNET_CADET_Channel *channel,
1341                         void **channel_ctx,
1342                         const struct GNUNET_MessageHeader *msg)
1343 {
1344   struct GNUNET_RPS_P2P_PullReplyMessage *in_msg;
1345   struct GNUNET_PeerIdentity *peers;
1346   struct GNUNET_PeerIdentity *sender;
1347   uint32_t i;
1348 #ifdef ENABLE_MALICIOUS
1349   struct AttackedPeer *tmp_att_peer;
1350 #endif /* ENABLE_MALICIOUS */
1351
1352   /* Check for protocol violation */
1353   if (sizeof (struct GNUNET_RPS_P2P_PullReplyMessage) > ntohs (msg->size))
1354   {
1355     GNUNET_break_op (0);
1356     GNUNET_CADET_receive_done (channel);
1357     return GNUNET_SYSERR;
1358   }
1359
1360   in_msg = (struct GNUNET_RPS_P2P_PullReplyMessage *) msg;
1361   if ((ntohs (msg->size) - sizeof (struct GNUNET_RPS_P2P_PullReplyMessage)) /
1362       sizeof (struct GNUNET_PeerIdentity) != ntohl (in_msg->num_peers))
1363   {
1364     LOG (GNUNET_ERROR_TYPE_ERROR,
1365         "message says it sends %" PRIu32 " peers, have space for %lu peers\n",
1366         ntohl (in_msg->num_peers),
1367         (ntohs (msg->size) - sizeof (struct GNUNET_RPS_P2P_PullReplyMessage)) /
1368             sizeof (struct GNUNET_PeerIdentity));
1369     GNUNET_break_op (0);
1370     GNUNET_CADET_receive_done (channel);
1371     return GNUNET_SYSERR;
1372   }
1373
1374   // Guess simply casting isn't the nicest way...
1375   // FIXME wait for cadet to change this function
1376   sender = (struct GNUNET_PeerIdentity *)
1377       GNUNET_CADET_channel_get_info (channel, GNUNET_CADET_OPTION_PEER);
1378
1379   LOG (GNUNET_ERROR_TYPE_DEBUG, "Received PULL REPLY (%s)\n", GNUNET_i2s (sender));
1380
1381   if (GNUNET_YES != Peers_check_peer_flag (sender, Peers_PULL_REPLY_PENDING))
1382   {
1383     LOG (GNUNET_ERROR_TYPE_WARNING,
1384         "Received a pull reply from a peer we didn't request one from!\n");
1385     GNUNET_CADET_receive_done (channel);
1386     GNUNET_break_op (0);
1387     return GNUNET_OK;
1388   }
1389
1390
1391   #ifdef ENABLE_MALICIOUS
1392   // We shouldn't even receive pull replies as we're not sending
1393   if (2 == mal_type)
1394   {
1395     GNUNET_CADET_receive_done (channel);
1396     return GNUNET_OK;
1397   }
1398   #endif /* ENABLE_MALICIOUS */
1399
1400   /* Do actual logic */
1401   peers = (struct GNUNET_PeerIdentity *) &in_msg[1];
1402
1403   LOG (GNUNET_ERROR_TYPE_DEBUG,
1404        "PULL REPLY received, got following %u peers:\n",
1405        ntohl (in_msg->num_peers));
1406
1407   for (i = 0 ; i < ntohl (in_msg->num_peers) ; i++)
1408   {
1409     LOG (GNUNET_ERROR_TYPE_DEBUG,
1410          "%u. %s\n",
1411          i,
1412          GNUNET_i2s (&peers[i]));
1413
1414     #ifdef ENABLE_MALICIOUS
1415     if ((NULL != att_peer_set) &&
1416         (1 == mal_type || 3 == mal_type))
1417     { /* Add attacked peer to local list */
1418       // TODO check if we sent a request and this was the first reply
1419       if (GNUNET_NO == GNUNET_CONTAINER_multipeermap_contains (att_peer_set,
1420                                                                &peers[i])
1421           && GNUNET_NO == GNUNET_CONTAINER_multipeermap_contains (mal_peer_set,
1422                                                                   &peers[i])
1423           && 0 != GNUNET_CRYPTO_cmp_peer_identity (&peers[i],
1424                                                    &own_identity))
1425       {
1426         tmp_att_peer = GNUNET_new (struct AttackedPeer);
1427         tmp_att_peer->peer_id = peers[i];
1428         GNUNET_CONTAINER_DLL_insert (att_peers_head,
1429                                      att_peers_tail,
1430                                      tmp_att_peer);
1431         add_peer_array_to_set (&peers[i], 1, att_peer_set);
1432       }
1433       continue;
1434     }
1435     #endif /* ENABLE_MALICIOUS */
1436     if (0 != GNUNET_CRYPTO_cmp_peer_identity (&own_identity,
1437                                               &peers[i]))
1438     {
1439       /* Make sure we 'know' about this peer */
1440       (void) Peers_insert_peer (&peers[i]);
1441
1442       if (GNUNET_YES == Peers_check_peer_valid (&peers[i]))
1443       {
1444         CustomPeerMap_put (pull_map, &peers[i]);
1445       }
1446       else
1447       {
1448         Peers_schedule_operation (&peers[i], insert_in_pull_map);
1449         (void) Peers_issue_peer_liveliness_check (&peers[i]);
1450       }
1451     }
1452   }
1453
1454   Peers_unset_peer_flag (sender, Peers_PULL_REPLY_PENDING);
1455   clean_peer (sender);
1456
1457   GNUNET_CADET_receive_done (channel);
1458   return GNUNET_OK;
1459 }
1460
1461
1462 /**
1463  * Compute a random delay.
1464  * A uniformly distributed value between mean + spread and mean - spread.
1465  *
1466  * For example for mean 4 min and spread 2 the minimum is (4 min - (1/2 * 4 min))
1467  * It would return a random value between 2 and 6 min.
1468  *
1469  * @param mean the mean
1470  * @param spread the inverse amount of deviation from the mean
1471  */
1472 static struct GNUNET_TIME_Relative
1473 compute_rand_delay (struct GNUNET_TIME_Relative mean,
1474                     unsigned int spread)
1475 {
1476   struct GNUNET_TIME_Relative half_interval;
1477   struct GNUNET_TIME_Relative ret;
1478   unsigned int rand_delay;
1479   unsigned int max_rand_delay;
1480
1481   if (0 == spread)
1482   {
1483     LOG (GNUNET_ERROR_TYPE_WARNING,
1484          "Not accepting spread of 0\n");
1485     GNUNET_break (0);
1486     GNUNET_assert (0);
1487   }
1488   GNUNET_assert (0 != mean.rel_value_us);
1489
1490   /* Compute random time value between spread * mean and spread * mean */
1491   half_interval = GNUNET_TIME_relative_divide (mean, spread);
1492
1493   max_rand_delay = GNUNET_TIME_UNIT_FOREVER_REL.rel_value_us / mean.rel_value_us * (2/spread);
1494   /**
1495    * Compute random value between (0 and 1) * round_interval
1496    * via multiplying round_interval with a 'fraction' (0 to value)/value
1497    */
1498   rand_delay = GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK, max_rand_delay);
1499   ret = GNUNET_TIME_relative_multiply (mean,  rand_delay);
1500   ret = GNUNET_TIME_relative_divide   (ret, max_rand_delay);
1501   ret = GNUNET_TIME_relative_add      (ret, half_interval);
1502
1503   if (GNUNET_TIME_UNIT_FOREVER_REL.rel_value_us == ret.rel_value_us)
1504     LOG (GNUNET_ERROR_TYPE_WARNING,
1505          "Returning FOREVER_REL\n");
1506
1507   return ret;
1508 }
1509
1510
1511 /**
1512  * Send single pull request
1513  *
1514  * @param peer_id the peer to send the pull request to.
1515  */
1516 static void
1517 send_pull_request (const struct GNUNET_PeerIdentity *peer)
1518 {
1519   struct GNUNET_MQ_Envelope *ev;
1520
1521   GNUNET_assert (GNUNET_NO == Peers_check_peer_flag (peer,
1522                                                      Peers_PULL_REPLY_PENDING));
1523   Peers_set_peer_flag (peer, Peers_PULL_REPLY_PENDING);
1524
1525   LOG (GNUNET_ERROR_TYPE_DEBUG,
1526        "Going to send PULL REQUEST to peer %s.\n",
1527        GNUNET_i2s (peer));
1528
1529   ev = GNUNET_MQ_msg_header (GNUNET_MESSAGE_TYPE_RPS_PP_PULL_REQUEST);
1530   Peers_send_message (peer, ev, "PULL REQUEST");
1531 }
1532
1533
1534 /**
1535  * Send single push
1536  *
1537  * @param peer_id the peer to send the push to.
1538  */
1539 static void
1540 send_push (const struct GNUNET_PeerIdentity *peer_id)
1541 {
1542   struct GNUNET_MQ_Envelope *ev;
1543
1544   LOG (GNUNET_ERROR_TYPE_DEBUG,
1545        "Going to send PUSH to peer %s.\n",
1546        GNUNET_i2s (peer_id));
1547
1548   ev = GNUNET_MQ_msg_header (GNUNET_MESSAGE_TYPE_RPS_PP_PUSH);
1549   Peers_send_message (peer_id, ev, "PUSH");
1550 }
1551
1552
1553 static void
1554 do_round (void *cls);
1555
1556 static void
1557 do_mal_round (void *cls);
1558
1559
1560 #ifdef ENABLE_MALICIOUS
1561 /**
1562  * Turn RPS service to act malicious.
1563  *
1564  * @param cls Closure
1565  * @param client The client that sent the message
1566  * @param msg The message header
1567  */
1568 static void
1569 handle_client_act_malicious (void *cls,
1570                              const struct GNUNET_MessageHeader *msg)
1571 {
1572   struct ClientContext *cli_ctx = cls;
1573   struct GNUNET_RPS_CS_ActMaliciousMessage *in_msg;
1574   struct GNUNET_PeerIdentity *peers;
1575   uint32_t num_mal_peers_sent;
1576   uint32_t num_mal_peers_old;
1577
1578   /* Check for protocol violation */
1579   if (sizeof (struct GNUNET_RPS_CS_ActMaliciousMessage) > ntohs (msg->size))
1580   {
1581     GNUNET_SERVICE_client_continue (cli_ctx->client);
1582     GNUNET_break_op (0);
1583   }
1584
1585   in_msg = (struct GNUNET_RPS_CS_ActMaliciousMessage *) msg;
1586   if ((ntohs (msg->size) - sizeof (struct GNUNET_RPS_CS_ActMaliciousMessage)) /
1587       sizeof (struct GNUNET_PeerIdentity) != ntohl (in_msg->num_peers))
1588   {
1589     LOG (GNUNET_ERROR_TYPE_ERROR,
1590         "message says it sends %" PRIu32 " peers, have space for %lu peers\n",
1591         ntohl (in_msg->num_peers),
1592         (ntohs (msg->size) - sizeof (struct GNUNET_RPS_CS_ActMaliciousMessage)) /
1593             sizeof (struct GNUNET_PeerIdentity));
1594     GNUNET_SERVICE_client_continue (cli_ctx->client);
1595     GNUNET_break_op (0);
1596   }
1597
1598
1599   /* Do actual logic */
1600   peers = (struct GNUNET_PeerIdentity *) &msg[1];
1601   mal_type = ntohl (in_msg->type);
1602   if (NULL == mal_peer_set)
1603     mal_peer_set = GNUNET_CONTAINER_multipeermap_create (1, GNUNET_NO);
1604
1605   LOG (GNUNET_ERROR_TYPE_DEBUG,
1606        "Now acting malicious type %" PRIu32 ", got %" PRIu32 " peers.\n",
1607        mal_type,
1608        ntohl (in_msg->num_peers));
1609
1610   if (1 == mal_type)
1611   { /* Try to maximise representation */
1612     /* Add other malicious peers to those we already know */
1613
1614     num_mal_peers_sent = ntohl (in_msg->num_peers);
1615     num_mal_peers_old = num_mal_peers;
1616     GNUNET_array_grow (mal_peers,
1617                        num_mal_peers,
1618                        num_mal_peers + num_mal_peers_sent);
1619     GNUNET_memcpy (&mal_peers[num_mal_peers_old],
1620             peers,
1621             num_mal_peers_sent * sizeof (struct GNUNET_PeerIdentity));
1622
1623     /* Add all mal peers to mal_peer_set */
1624     add_peer_array_to_set (&mal_peers[num_mal_peers_old],
1625                            num_mal_peers_sent,
1626                            mal_peer_set);
1627
1628     /* Substitute do_round () with do_mal_round () */
1629     GNUNET_SCHEDULER_cancel (do_round_task);
1630     do_round_task = GNUNET_SCHEDULER_add_now (&do_mal_round, NULL);
1631   }
1632
1633   else if ( (2 == mal_type) ||
1634             (3 == mal_type) )
1635   { /* Try to partition the network */
1636     /* Add other malicious peers to those we already know */
1637
1638     num_mal_peers_sent = ntohl (in_msg->num_peers) - 1;
1639     num_mal_peers_old = num_mal_peers;
1640     GNUNET_array_grow (mal_peers,
1641                        num_mal_peers,
1642                        num_mal_peers + num_mal_peers_sent);
1643     if (NULL != mal_peers &&
1644         0 != num_mal_peers)
1645     {
1646       GNUNET_memcpy (&mal_peers[num_mal_peers_old],
1647               peers,
1648               num_mal_peers_sent * sizeof (struct GNUNET_PeerIdentity));
1649
1650       /* Add all mal peers to mal_peer_set */
1651       add_peer_array_to_set (&mal_peers[num_mal_peers_old],
1652                              num_mal_peers_sent,
1653                              mal_peer_set);
1654     }
1655
1656     /* Store the one attacked peer */
1657     GNUNET_memcpy (&attacked_peer,
1658             &in_msg->attacked_peer,
1659             sizeof (struct GNUNET_PeerIdentity));
1660     /* Set the flag of the attacked peer to valid to avoid problems */
1661     if (GNUNET_NO == Peers_check_peer_known (&attacked_peer))
1662     {
1663       (void) Peers_issue_peer_liveliness_check (&attacked_peer);
1664     }
1665
1666     LOG (GNUNET_ERROR_TYPE_DEBUG,
1667          "Attacked peer is %s\n",
1668          GNUNET_i2s (&attacked_peer));
1669
1670     /* Substitute do_round () with do_mal_round () */
1671     GNUNET_SCHEDULER_cancel (do_round_task);
1672     do_round_task = GNUNET_SCHEDULER_add_now (&do_mal_round, NULL);
1673   }
1674   else if (0 == mal_type)
1675   { /* Stop acting malicious */
1676     GNUNET_array_grow (mal_peers, num_mal_peers, 0);
1677
1678     /* Substitute do_mal_round () with do_round () */
1679     GNUNET_SCHEDULER_cancel (do_round_task);
1680     do_round_task = GNUNET_SCHEDULER_add_now (&do_round, NULL);
1681   }
1682   else
1683   {
1684     GNUNET_break (0);
1685     GNUNET_SERVICE_client_continue (cli_ctx->client);
1686   }
1687   GNUNET_SERVICE_client_continue (cli_ctx->client);
1688 }
1689
1690
1691 /**
1692  * Send out PUSHes and PULLs maliciously.
1693  *
1694  * This is executed regylary.
1695  */
1696 static void
1697 do_mal_round (void *cls)
1698 {
1699   uint32_t num_pushes;
1700   uint32_t i;
1701   struct GNUNET_TIME_Relative time_next_round;
1702   struct AttackedPeer *tmp_att_peer;
1703
1704   LOG (GNUNET_ERROR_TYPE_DEBUG,
1705        "Going to execute next round maliciously type %" PRIu32 ".\n",
1706       mal_type);
1707   do_round_task = NULL;
1708   GNUNET_assert (mal_type <= 3);
1709   /* Do malicious actions */
1710   if (1 == mal_type)
1711   { /* Try to maximise representation */
1712
1713     /* The maximum of pushes we're going to send this round */
1714     num_pushes = GNUNET_MIN (GNUNET_MIN (push_limit,
1715                                          num_attacked_peers),
1716                              GNUNET_CONSTANTS_MAX_CADET_MESSAGE_SIZE);
1717
1718     LOG (GNUNET_ERROR_TYPE_DEBUG,
1719          "Going to send %" PRIu32 " pushes\n",
1720          num_pushes);
1721
1722     /* Send PUSHes to attacked peers */
1723     for (i = 0 ; i < num_pushes ; i++)
1724     {
1725       if (att_peers_tail == att_peer_index)
1726         att_peer_index = att_peers_head;
1727       else
1728         att_peer_index = att_peer_index->next;
1729
1730       send_push (&att_peer_index->peer_id);
1731     }
1732
1733     /* Send PULLs to some peers to learn about additional peers to attack */
1734     tmp_att_peer = att_peer_index;
1735     for (i = 0 ; i < num_pushes * alpha ; i++)
1736     {
1737       if (att_peers_tail == tmp_att_peer)
1738         tmp_att_peer = att_peers_head;
1739       else
1740         att_peer_index = tmp_att_peer->next;
1741
1742       send_pull_request (&tmp_att_peer->peer_id);
1743     }
1744   }
1745
1746
1747   else if (2 == mal_type)
1748   { /**
1749      * Try to partition the network
1750      * Send as many pushes to the attacked peer as possible
1751      * That is one push per round as it will ignore more.
1752      */
1753     (void) Peers_issue_peer_liveliness_check (&attacked_peer);
1754     if (GNUNET_YES == Peers_check_peer_flag (&attacked_peer, Peers_ONLINE))
1755       send_push (&attacked_peer);
1756   }
1757
1758
1759   if (3 == mal_type)
1760   { /* Combined attack */
1761
1762     /* Send PUSH to attacked peers */
1763     if (GNUNET_YES == Peers_check_peer_known (&attacked_peer))
1764     {
1765       (void) Peers_issue_peer_liveliness_check (&attacked_peer);
1766       if (GNUNET_YES == Peers_check_peer_flag (&attacked_peer, Peers_ONLINE))
1767       {
1768         LOG (GNUNET_ERROR_TYPE_DEBUG,
1769             "Goding to send push to attacked peer (%s)\n",
1770             GNUNET_i2s (&attacked_peer));
1771         send_push (&attacked_peer);
1772       }
1773     }
1774     (void) Peers_issue_peer_liveliness_check (&attacked_peer);
1775
1776     /* The maximum of pushes we're going to send this round */
1777     num_pushes = GNUNET_MIN (GNUNET_MIN (push_limit - 1,
1778                                          num_attacked_peers),
1779                              GNUNET_CONSTANTS_MAX_CADET_MESSAGE_SIZE);
1780
1781     LOG (GNUNET_ERROR_TYPE_DEBUG,
1782          "Going to send %" PRIu32 " pushes\n",
1783          num_pushes);
1784
1785     for (i = 0; i < num_pushes; i++)
1786     {
1787       if (att_peers_tail == att_peer_index)
1788         att_peer_index = att_peers_head;
1789       else
1790         att_peer_index = att_peer_index->next;
1791
1792       send_push (&att_peer_index->peer_id);
1793     }
1794
1795     /* Send PULLs to some peers to learn about additional peers to attack */
1796     tmp_att_peer = att_peer_index;
1797     for (i = 0; i < num_pushes * alpha; i++)
1798     {
1799       if (att_peers_tail == tmp_att_peer)
1800         tmp_att_peer = att_peers_head;
1801       else
1802         att_peer_index = tmp_att_peer->next;
1803
1804       send_pull_request (&tmp_att_peer->peer_id);
1805     }
1806   }
1807
1808   /* Schedule next round */
1809   time_next_round = compute_rand_delay (round_interval, 2);
1810
1811   //do_round_task = GNUNET_SCHEDULER_add_delayed (round_interval, &do_mal_round,
1812   //NULL);
1813   GNUNET_assert (NULL == do_round_task);
1814   do_round_task = GNUNET_SCHEDULER_add_delayed (time_next_round,
1815                                                 &do_mal_round, NULL);
1816   LOG (GNUNET_ERROR_TYPE_DEBUG, "Finished round\n");
1817 }
1818 #endif /* ENABLE_MALICIOUS */
1819
1820
1821 /**
1822  * Send out PUSHes and PULLs, possibly update #view, samplers.
1823  *
1824  * This is executed regylary.
1825  */
1826 static void
1827 do_round (void *cls)
1828 {
1829   uint32_t i;
1830   const struct GNUNET_PeerIdentity *view_array;
1831   unsigned int *permut;
1832   unsigned int a_peers; /* Number of peers we send pushes to */
1833   unsigned int b_peers; /* Number of peers we send pull requests to */
1834   uint32_t first_border;
1835   uint32_t second_border;
1836   struct GNUNET_PeerIdentity peer;
1837   struct GNUNET_PeerIdentity *update_peer;
1838
1839   LOG (GNUNET_ERROR_TYPE_DEBUG,
1840        "Going to execute next round.\n");
1841   do_round_task = NULL;
1842   LOG (GNUNET_ERROR_TYPE_DEBUG,
1843        "Printing view:\n");
1844   to_file (file_name_view_log,
1845            "___ new round ___");
1846   view_array = View_get_as_array ();
1847   for (i = 0; i < View_size (); i++)
1848   {
1849     LOG (GNUNET_ERROR_TYPE_DEBUG,
1850          "\t%s\n", GNUNET_i2s (&view_array[i]));
1851     to_file (file_name_view_log,
1852              "=%s\t(do round)",
1853              GNUNET_i2s_full (&view_array[i]));
1854   }
1855
1856
1857   /* Send pushes and pull requests */
1858   if (0 < View_size ())
1859   {
1860     permut = GNUNET_CRYPTO_random_permute (GNUNET_CRYPTO_QUALITY_STRONG,
1861                                            View_size ());
1862
1863     /* Send PUSHes */
1864     a_peers = ceil (alpha * View_size ());
1865
1866     LOG (GNUNET_ERROR_TYPE_DEBUG,
1867          "Going to send pushes to %u (ceil (%f * %u)) peers.\n",
1868          a_peers, alpha, View_size ());
1869     for (i = 0; i < a_peers; i++)
1870     {
1871       peer = view_array[permut[i]];
1872       if (0 != GNUNET_CRYPTO_cmp_peer_identity (&own_identity, &peer)) // TODO
1873       { // FIXME if this fails schedule/loop this for later
1874         send_push (&peer);
1875       }
1876     }
1877
1878     /* Send PULL requests */
1879     b_peers = ceil (beta * View_size ());
1880     first_border = a_peers;
1881     second_border = a_peers + b_peers;
1882     if (second_border > View_size ())
1883     {
1884       first_border = View_size () - b_peers;
1885       second_border = View_size ();
1886     }
1887     LOG (GNUNET_ERROR_TYPE_DEBUG,
1888         "Going to send pulls to %u (ceil (%f * %u)) peers.\n",
1889         b_peers, beta, View_size ());
1890     for (i = first_border; i < second_border; i++)
1891     {
1892       peer = view_array[permut[i]];
1893       if (0 != GNUNET_CRYPTO_cmp_peer_identity (&own_identity, &peer) &&
1894           GNUNET_NO == Peers_check_peer_flag (&peer, Peers_PULL_REPLY_PENDING)) // TODO
1895       { // FIXME if this fails schedule/loop this for later
1896         send_pull_request (&peer);
1897       }
1898     }
1899
1900     GNUNET_free (permut);
1901     permut = NULL;
1902   }
1903
1904
1905   /* Update view */
1906   /* TODO see how many peers are in push-/pull- list! */
1907
1908   if ((CustomPeerMap_size (push_map) <= alpha * View_size ()) &&
1909       (0 < CustomPeerMap_size (push_map)) &&
1910       (0 < CustomPeerMap_size (pull_map)))
1911   { /* If conditions for update are fulfilled, update */
1912     LOG (GNUNET_ERROR_TYPE_DEBUG, "Update of the view.\n");
1913
1914     uint32_t final_size;
1915     uint32_t peers_to_clean_size;
1916     struct GNUNET_PeerIdentity *peers_to_clean;
1917
1918     peers_to_clean = NULL;
1919     peers_to_clean_size = 0;
1920     GNUNET_array_grow (peers_to_clean, peers_to_clean_size, View_size ());
1921     GNUNET_memcpy (peers_to_clean,
1922             view_array,
1923             View_size () * sizeof (struct GNUNET_PeerIdentity));
1924
1925     /* Seems like recreating is the easiest way of emptying the peermap */
1926     View_clear ();
1927     to_file (file_name_view_log,
1928              "--- emptied ---");
1929
1930     first_border  = GNUNET_MIN (ceil (alpha * sampler_size_est_need),
1931                                 CustomPeerMap_size (push_map));
1932     second_border = first_border +
1933                     GNUNET_MIN (floor (beta  * sampler_size_est_need),
1934                                 CustomPeerMap_size (pull_map));
1935     final_size    = second_border +
1936       ceil ((1 - (alpha + beta)) * sampler_size_est_need);
1937
1938     /* Update view with peers received through PUSHes */
1939     permut = GNUNET_CRYPTO_random_permute (GNUNET_CRYPTO_QUALITY_STRONG,
1940                                            CustomPeerMap_size (push_map));
1941     for (i = 0; i < first_border; i++)
1942     {
1943       (void) insert_in_view (CustomPeerMap_get_peer_by_index (push_map,
1944                                                               permut[i]));
1945       to_file (file_name_view_log,
1946                "+%s\t(push list)",
1947                GNUNET_i2s_full (&view_array[i]));
1948       // TODO change the peer_flags accordingly
1949     }
1950     GNUNET_free (permut);
1951     permut = NULL;
1952
1953     /* Update view with peers received through PULLs */
1954     permut = GNUNET_CRYPTO_random_permute (GNUNET_CRYPTO_QUALITY_STRONG,
1955                                            CustomPeerMap_size (pull_map));
1956     for (i = first_border; i < second_border; i++)
1957     {
1958       (void) insert_in_view (CustomPeerMap_get_peer_by_index (pull_map,
1959             permut[i - first_border]));
1960       to_file (file_name_view_log,
1961                "+%s\t(pull list)",
1962                GNUNET_i2s_full (&view_array[i]));
1963       // TODO change the peer_flags accordingly
1964     }
1965     GNUNET_free (permut);
1966     permut = NULL;
1967
1968     /* Update view with peers from history */
1969     RPS_sampler_get_n_rand_peers (prot_sampler,
1970                                   hist_update,
1971                                   NULL,
1972                                   final_size - second_border);
1973     // TODO change the peer_flags accordingly
1974
1975     for (i = 0; i < View_size (); i++)
1976       rem_from_list (&peers_to_clean, &peers_to_clean_size, &view_array[i]);
1977
1978     /* Clean peers that were removed from the view */
1979     for (i = 0; i < peers_to_clean_size; i++)
1980     {
1981       to_file (file_name_view_log,
1982                "-%s",
1983                GNUNET_i2s_full (&peers_to_clean[i]));
1984       clean_peer (&peers_to_clean[i]);
1985       //peer_destroy_channel_send (sender);
1986     }
1987
1988     GNUNET_array_grow (peers_to_clean, peers_to_clean_size, 0);
1989   }
1990   else
1991   {
1992     LOG (GNUNET_ERROR_TYPE_DEBUG, "No update of the view.\n");
1993   }
1994   // TODO independent of that also get some peers from CADET_get_peers()?
1995
1996   LOG (GNUNET_ERROR_TYPE_DEBUG,
1997        "Received %u pushes and %u pulls last round (alpha (%.2f) * view_size (%u) = %.2f)\n",
1998        CustomPeerMap_size (push_map),
1999        CustomPeerMap_size (pull_map),
2000        alpha,
2001        View_size (),
2002        alpha * View_size ());
2003
2004   /* Update samplers */
2005   for (i = 0; i < CustomPeerMap_size (push_map); i++)
2006   {
2007     update_peer = CustomPeerMap_get_peer_by_index (push_map, i);
2008     LOG (GNUNET_ERROR_TYPE_DEBUG,
2009          "Updating with peer %s from push list\n",
2010          GNUNET_i2s (update_peer));
2011     insert_in_sampler (NULL, update_peer);
2012     clean_peer (update_peer); /* This cleans only if it is not in the view */
2013     //peer_destroy_channel_send (sender);
2014   }
2015
2016   for (i = 0; i < CustomPeerMap_size (pull_map); i++)
2017   {
2018     LOG (GNUNET_ERROR_TYPE_DEBUG,
2019          "Updating with peer %s from pull list\n",
2020          GNUNET_i2s (CustomPeerMap_get_peer_by_index (pull_map, i)));
2021     insert_in_sampler (NULL, CustomPeerMap_get_peer_by_index (pull_map, i));
2022     /* This cleans only if it is not in the view */
2023     clean_peer (CustomPeerMap_get_peer_by_index (pull_map, i));
2024     //peer_destroy_channel_send (sender);
2025   }
2026
2027
2028   /* Empty push/pull lists */
2029   CustomPeerMap_clear (push_map);
2030   CustomPeerMap_clear (pull_map);
2031
2032   struct GNUNET_TIME_Relative time_next_round;
2033
2034   time_next_round = compute_rand_delay (round_interval, 2);
2035
2036   /* Schedule next round */
2037   do_round_task = GNUNET_SCHEDULER_add_delayed (time_next_round,
2038                                                 &do_round, NULL);
2039   LOG (GNUNET_ERROR_TYPE_DEBUG, "Finished round\n");
2040 }
2041
2042
2043 /**
2044  * This is called from GNUNET_CADET_get_peers().
2045  *
2046  * It is called on every peer(ID) that cadet somehow has contact with.
2047  * We use those to initialise the sampler.
2048  */
2049 void
2050 init_peer_cb (void *cls,
2051               const struct GNUNET_PeerIdentity *peer,
2052               int tunnel, // "Do we have a tunnel towards this peer?"
2053               unsigned int n_paths, // "Number of known paths towards this peer"
2054               unsigned int best_path) // "How long is the best path?
2055                                       // (0 = unknown, 1 = ourselves, 2 = neighbor)"
2056 {
2057   if (NULL != peer)
2058   {
2059     LOG (GNUNET_ERROR_TYPE_DEBUG,
2060          "Got peer_id %s from cadet\n",
2061          GNUNET_i2s (peer));
2062     got_peer (peer);
2063   }
2064 }
2065
2066 /**
2067  * @brief Iterator function over stored, valid peers.
2068  *
2069  * We initialise the sampler with those.
2070  *
2071  * @param cls the closure
2072  * @param peer the peer id
2073  * @return #GNUNET_YES if we should continue to
2074  *         iterate,
2075  *         #GNUNET_NO if not.
2076  */
2077 static int
2078 valid_peers_iterator (void *cls,
2079                       const struct GNUNET_PeerIdentity *peer)
2080 {
2081   if (NULL != peer)
2082   {
2083     LOG (GNUNET_ERROR_TYPE_DEBUG,
2084          "Got stored, valid peer %s\n",
2085          GNUNET_i2s (peer));
2086     got_peer (peer);
2087   }
2088   return GNUNET_YES;
2089 }
2090
2091
2092 /**
2093  * Iterator over peers from peerinfo.
2094  *
2095  * @param cls closure
2096  * @param peer id of the peer, NULL for last call
2097  * @param hello hello message for the peer (can be NULL)
2098  * @param error message
2099  */
2100 void
2101 process_peerinfo_peers (void *cls,
2102                         const struct GNUNET_PeerIdentity *peer,
2103                         const struct GNUNET_HELLO_Message *hello,
2104                         const char *err_msg)
2105 {
2106   if (NULL != peer)
2107   {
2108     LOG (GNUNET_ERROR_TYPE_DEBUG,
2109          "Got peer_id %s from peerinfo\n",
2110          GNUNET_i2s (peer));
2111     got_peer (peer);
2112   }
2113 }
2114
2115
2116 /**
2117  * Task run during shutdown.
2118  *
2119  * @param cls unused
2120  */
2121 static void
2122 shutdown_task (void *cls)
2123 {
2124   struct ClientContext *client_ctx;
2125   struct ReplyCls *reply_cls;
2126
2127   LOG (GNUNET_ERROR_TYPE_DEBUG,
2128        "RPS is going down\n");
2129
2130   /* Clean all clients */
2131   for (client_ctx = cli_ctx_head;
2132        NULL != cli_ctx_head;
2133        client_ctx = cli_ctx_head)
2134   {
2135     /* Clean pending requests to the sampler */
2136     for (reply_cls = client_ctx->rep_cls_head;
2137          NULL != client_ctx->rep_cls_head;
2138          reply_cls = client_ctx->rep_cls_head)
2139     {
2140       RPS_sampler_request_cancel (reply_cls->req_handle);
2141       GNUNET_CONTAINER_DLL_remove (client_ctx->rep_cls_head,
2142                                    client_ctx->rep_cls_tail,
2143                                    reply_cls);
2144       GNUNET_free (reply_cls);
2145     }
2146     GNUNET_MQ_destroy (client_ctx->mq);
2147     GNUNET_CONTAINER_DLL_remove (cli_ctx_head, cli_ctx_tail, client_ctx);
2148     GNUNET_free (client_ctx);
2149   }
2150   GNUNET_PEERINFO_notify_cancel (peerinfo_notify_handle);
2151   GNUNET_PEERINFO_disconnect (peerinfo_handle);
2152
2153   if (NULL != do_round_task)
2154   {
2155     GNUNET_SCHEDULER_cancel (do_round_task);
2156     do_round_task = NULL;
2157   }
2158
2159   Peers_terminate ();
2160
2161   GNUNET_NSE_disconnect (nse);
2162   RPS_sampler_destroy (prot_sampler);
2163   RPS_sampler_destroy (client_sampler);
2164   GNUNET_CADET_disconnect (cadet_handle);
2165   View_destroy ();
2166   CustomPeerMap_destroy (push_map);
2167   CustomPeerMap_destroy (pull_map);
2168   #ifdef ENABLE_MALICIOUS
2169   struct AttackedPeer *tmp_att_peer;
2170   GNUNET_free (file_name_view_log);
2171   GNUNET_array_grow (mal_peers, num_mal_peers, 0);
2172   if (NULL != mal_peer_set)
2173     GNUNET_CONTAINER_multipeermap_destroy (mal_peer_set);
2174   if (NULL != att_peer_set)
2175     GNUNET_CONTAINER_multipeermap_destroy (att_peer_set);
2176   while (NULL != att_peers_head)
2177   {
2178     tmp_att_peer = att_peers_head;
2179     GNUNET_CONTAINER_DLL_remove (att_peers_head, att_peers_tail, tmp_att_peer);
2180   }
2181   #endif /* ENABLE_MALICIOUS */
2182 }
2183
2184
2185 /**
2186  * Handle client connecting to the service.
2187  *
2188  * @param cls NULL
2189  * @param client the new client
2190  * @param mq the message queue of @a client
2191  * @return @a client
2192  */
2193 static void *
2194 client_connect_cb (void *cls,
2195                    struct GNUNET_SERVICE_Client *client,
2196                    struct GNUNET_MQ_Handle *mq)
2197 {
2198   struct ClientContext *cli_ctx;
2199
2200   LOG (GNUNET_ERROR_TYPE_DEBUG,
2201        "Client connected\n");
2202   if (NULL == client)
2203     return client; /* Server was destroyed before a client connected. Shutting down */
2204   cli_ctx = GNUNET_new (struct ClientContext);
2205   cli_ctx->mq = GNUNET_SERVICE_client_get_mq (client);
2206   cli_ctx->client = client;
2207   GNUNET_CONTAINER_DLL_insert (cli_ctx_head,
2208                                cli_ctx_tail,
2209                                cli_ctx);
2210   return cli_ctx;
2211 }
2212
2213 /**
2214  * Callback called when a client disconnected from the service
2215  *
2216  * @param cls closure for the service
2217  * @param c the client that disconnected
2218  * @param internal_cls should be equal to @a c
2219  */
2220 static void
2221 client_disconnect_cb (void *cls,
2222                                   struct GNUNET_SERVICE_Client *client,
2223                                   void *internal_cls)
2224 {
2225   struct ClientContext *cli_ctx = internal_cls;
2226
2227   GNUNET_assert (client == cli_ctx->client);
2228   if (NULL == client)
2229   {/* shutdown task */
2230     while (NULL != cli_ctx_head)
2231       destroy_cli_ctx (cli_ctx_head);
2232   }
2233   else
2234   {
2235     destroy_cli_ctx (cli_ctx);
2236   }
2237 }
2238
2239
2240 /**
2241  * Handle random peer sampling clients.
2242  *
2243  * @param cls closure
2244  * @param c configuration to use
2245  * @param service the initialized service
2246  */
2247 static void
2248 run (void *cls,
2249      const struct GNUNET_CONFIGURATION_Handle *c,
2250      struct GNUNET_SERVICE_Handle *service)
2251 {
2252   static const struct GNUNET_CADET_MessageHandler cadet_handlers[] = {
2253     {&handle_peer_check       , GNUNET_MESSAGE_TYPE_RPS_PP_CHECK_LIVE,
2254       sizeof (struct GNUNET_MessageHeader)},
2255     {&handle_peer_push        , GNUNET_MESSAGE_TYPE_RPS_PP_PUSH,
2256       sizeof (struct GNUNET_MessageHeader)},
2257     {&handle_peer_pull_request, GNUNET_MESSAGE_TYPE_RPS_PP_PULL_REQUEST,
2258       sizeof (struct GNUNET_MessageHeader)},
2259     {&handle_peer_pull_reply  , GNUNET_MESSAGE_TYPE_RPS_PP_PULL_REPLY, 0},
2260     {NULL, 0, 0}
2261   };
2262
2263   int size;
2264   int out_size;
2265   char* fn_valid_peers;
2266   struct GNUNET_HashCode port;
2267
2268   GNUNET_log_setup ("rps", GNUNET_error_type_to_string (GNUNET_ERROR_TYPE_DEBUG), NULL);
2269   cfg = c;
2270
2271
2272   /* Get own ID */
2273   GNUNET_CRYPTO_get_peer_identity (cfg, &own_identity); // TODO check return value
2274   GNUNET_log (GNUNET_ERROR_TYPE_INFO,
2275               "STARTING SERVICE (rps) for peer [%s]\n",
2276               GNUNET_i2s (&own_identity));
2277   #ifdef ENABLE_MALICIOUS
2278   GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
2279               "Malicious execution compiled in.\n");
2280   #endif /* ENABLE_MALICIOUS */
2281
2282
2283
2284   /* Get time interval from the configuration */
2285   if (GNUNET_OK != GNUNET_CONFIGURATION_get_value_time (cfg, "RPS",
2286                                                         "ROUNDINTERVAL",
2287                                                         &round_interval))
2288   {
2289     GNUNET_log_config_missing (GNUNET_ERROR_TYPE_ERROR,
2290                                "RPS", "ROUNDINTERVAL");
2291     GNUNET_SCHEDULER_shutdown ();
2292     return;
2293   }
2294
2295   /* Get initial size of sampler/view from the configuration */
2296   if (GNUNET_OK !=
2297       GNUNET_CONFIGURATION_get_value_number (cfg, "RPS", "INITSIZE",
2298         (long long unsigned int *) &sampler_size_est_need))
2299   {
2300     GNUNET_log_config_missing (GNUNET_ERROR_TYPE_ERROR,
2301                                "RPS", "INITSIZE");
2302     GNUNET_SCHEDULER_shutdown ();
2303     return;
2304   }
2305   LOG (GNUNET_ERROR_TYPE_DEBUG, "INITSIZE is %u\n", sampler_size_est_need);
2306
2307   if (GNUNET_OK !=
2308       GNUNET_CONFIGURATION_get_value_filename (cfg,
2309                                                "rps",
2310                                                "FILENAME_VALID_PEERS",
2311                                                &fn_valid_peers))
2312   {
2313     GNUNET_log_config_missing (GNUNET_ERROR_TYPE_ERROR,
2314                                "rps", "FILENAME_VALID_PEERS");
2315   }
2316
2317
2318   View_create (4);
2319
2320   /* file_name_view_log */
2321   if (GNUNET_OK != GNUNET_DISK_directory_create ("/tmp/rps/"))
2322   {
2323     LOG (GNUNET_ERROR_TYPE_WARNING,
2324          "Failed to create directory /tmp/rps/\n");
2325   }
2326
2327   size = (14 + strlen (GNUNET_i2s_full (&own_identity)) + 1) * sizeof (char);
2328   file_name_view_log = GNUNET_malloc (size);
2329   out_size = GNUNET_snprintf (file_name_view_log,
2330                               size,
2331                               "/tmp/rps/view-%s",
2332                               GNUNET_i2s_full (&own_identity));
2333   if (size < out_size ||
2334       0 > out_size)
2335   {
2336     LOG (GNUNET_ERROR_TYPE_WARNING,
2337          "Failed to write string to buffer (size: %i, out_size: %i)\n",
2338          size,
2339          out_size);
2340   }
2341
2342
2343   /* connect to NSE */
2344   nse = GNUNET_NSE_connect (cfg, nse_callback, NULL);
2345
2346
2347   alpha = 0.45;
2348   beta  = 0.45;
2349
2350
2351   /* Initialise cadet */
2352   cadet_handle = GNUNET_CADET_connect (cfg,
2353                                        cls,
2354                                        &cleanup_destroyed_channel,
2355                                        cadet_handlers);
2356   GNUNET_assert (NULL != cadet_handle);
2357   GNUNET_CRYPTO_hash (GNUNET_APPLICATION_PORT_RPS,
2358                       strlen (GNUNET_APPLICATION_PORT_RPS),
2359                       &port);
2360   GNUNET_CADET_open_port (cadet_handle,
2361                           &port,
2362                           &Peers_handle_inbound_channel, cls);
2363
2364
2365   peerinfo_handle = GNUNET_PEERINFO_connect (cfg);
2366   Peers_initialise (fn_valid_peers, cadet_handle, &own_identity);
2367   GNUNET_free (fn_valid_peers);
2368
2369   /* Initialise sampler */
2370   struct GNUNET_TIME_Relative half_round_interval;
2371   struct GNUNET_TIME_Relative  max_round_interval;
2372
2373   half_round_interval = GNUNET_TIME_relative_multiply (round_interval, .5);
2374   max_round_interval = GNUNET_TIME_relative_add (round_interval, half_round_interval);
2375
2376   prot_sampler =   RPS_sampler_init     (sampler_size_est_need, max_round_interval);
2377   client_sampler = RPS_sampler_mod_init (sampler_size_est_need, max_round_interval);
2378
2379   /* Initialise push and pull maps */
2380   push_map = CustomPeerMap_create (4);
2381   pull_map = CustomPeerMap_create (4);
2382
2383
2384   //LOG (GNUNET_ERROR_TYPE_DEBUG, "Requesting peers from CADET\n");
2385   //GNUNET_CADET_get_peers (cadet_handle, &init_peer_cb, NULL);
2386   // TODO send push/pull to each of those peers?
2387   // TODO read stored valid peers from last run
2388   LOG (GNUNET_ERROR_TYPE_DEBUG, "Requesting stored valid peers\n");
2389   Peers_get_valid_peers (valid_peers_iterator, NULL);
2390
2391   peerinfo_notify_handle = GNUNET_PEERINFO_notify (cfg,
2392                                                    GNUNET_NO,
2393                                                    process_peerinfo_peers,
2394                                                    NULL);
2395
2396   LOG (GNUNET_ERROR_TYPE_INFO, "Ready to receive requests from clients\n");
2397
2398   do_round_task = GNUNET_SCHEDULER_add_now (&do_round, NULL);
2399   LOG (GNUNET_ERROR_TYPE_DEBUG, "Scheduled first round\n");
2400
2401   GNUNET_SCHEDULER_add_shutdown (&shutdown_task, NULL);
2402 }
2403
2404
2405 /**
2406  * Define "main" method using service macro.
2407  */
2408 GNUNET_SERVICE_MAIN
2409 ("rps",
2410  GNUNET_SERVICE_OPTION_NONE,
2411  &run,
2412  &client_connect_cb,
2413  &client_disconnect_cb,
2414  NULL,
2415  GNUNET_MQ_hd_fixed_size (client_request,
2416    GNUNET_MESSAGE_TYPE_RPS_CS_REQUEST,
2417    struct GNUNET_RPS_CS_RequestMessage,
2418    NULL),
2419  GNUNET_MQ_hd_fixed_size (client_request_cancel,
2420    GNUNET_MESSAGE_TYPE_RPS_CS_REQUEST_CANCEL,
2421    struct GNUNET_RPS_CS_RequestCancelMessage,
2422    NULL),
2423  GNUNET_MQ_hd_fixed_size (client_seed,
2424    GNUNET_MESSAGE_TYPE_RPS_CS_SEED,
2425    struct GNUNET_MessageHeader,
2426    NULL),
2427 #ifdef ENABLE_MALICIOUS
2428  GNUNET_MQ_hd_fixed_size (client_act_malicious,
2429    GNUNET_MESSAGE_TYPE_RPS_ACT_MALICIOUS,
2430    struct GNUNET_MessageHeader,
2431    NULL),
2432 #endif /* ENABLE_MALICIOUS */
2433  GNUNET_MQ_handler_end());
2434
2435 /* end of gnunet-service-rps.c */