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