src/rps/gnunet-service-rps.c

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