src/rps/gnunet-service-rps.c

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