src/rps/gnunet-service-rps.c

   1 /*
   2      This file is part of GNUnet.
   3      (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 take care that messages are not longer than 64k
  42
  43 // TODO check for overflows
  44
  45 // TODO align message structs
  46
  47 // (TODO api -- possibility of getting weak random peer immideately)
  48
  49 // TODO malicious peer
  50
  51 // TODO Change API to accept initialisation peers
  52
  53 // TODO Change API to accept good peers 'friends'
  54
  55 // TODO store peers somewhere
  56
  57 // TODO check that every id we get is valid - is it reachable?
  58
  59 // TODO ignore list?
  60
  61 // hist_size_init, hist_size_max
  62
  63 /**
  64  * Our configuration.
  65  */
  66 static const struct GNUNET_CONFIGURATION_Handle *cfg;
  67
  68 /**
  69  * Our own identity.
  70  */
  71 static struct GNUNET_PeerIdentity *own_identity;
  72
  73
  74   struct GNUNET_PeerIdentity *
  75 get_rand_peer (const struct GNUNET_PeerIdentity *peer_list, unsigned int size);
  76
  77
  78 /***********************************************************************
  79  * Housekeeping with peers
  80 ***********************************************************************/
  81
  82 /**
  83  * Struct used to store the context of a connected client.
  84  */
  85 struct client_ctx
  86 {
  87   /**
  88    * The message queue to communicate with the client.
  89    */
  90   struct GNUNET_MQ_Handle *mq;
  91 };
  92
  93 /**
  94  * Used to keep track in what lists single peerIDs are.
  95  */
  96 enum PeerFlags
  97 {
  98   IN_OTHER_SAMPLER_LIST = 0x01, // unneeded?
  99   IN_OTHER_GOSSIP_LIST  = 0x02, // unneeded?
 100   IN_OWN_SAMPLER_LIST   = 0x04, // unneeded?
 101   IN_OWN_GOSSIP_LIST    = 0x08, // unneeded?
 102
 103   /**
 104    * We set this bit when we can be sure the other peer is/was live.
 105    */
 106   LIVING                = 0x10
 107 };
 108
 109
 110 /**
 111  * Functions of this type can be used to be stored at a peer for later execution.
 112  */
 113 typedef void (* PeerOp) (void *cls, const struct GNUNET_PeerIdentity *peer);
 114
 115 /**
 116  * Outstanding operation on peer consisting of callback and closure
 117  */
 118 struct PeerOutstandingOp
 119 {
 120   /**
 121    * Callback
 122    */
 123   PeerOp op;
 124
 125   /**
 126    * Closure
 127    */
 128   void *op_cls;
 129 };
 130
 131
 132 /**
 133  * Struct used to keep track of other peer's status
 134  *
 135  * This is stored in a multipeermap.
 136  */
 137 struct PeerContext
 138 {
 139   /**
 140    * In own gossip/sampler list, in other's gossip/sampler list
 141    */
 142   uint32_t peer_flags;
 143
 144   /**
 145    * Message queue open to client
 146    */
 147   struct GNUNET_MQ_Handle *mq;
 148
 149   /**
 150    * Channel open to client.
 151    */
 152   struct GNUNET_CADET_Channel *send_channel;
 153
 154   /**
 155    * Channel open from client.
 156    */
 157   struct GNUNET_CADET_Channel *recv_channel; // unneeded?
 158
 159   /**
 160    * Array of outstanding operations on this peer.
 161    */
 162   struct PeerOutstandingOp *outstanding_ops;
 163
 164   /**
 165    * Number of outstanding operations.
 166    */
 167   unsigned int num_outstanding_ops;
 168   //size_t num_outstanding_ops;
 169
 170   /**
 171    * This is pobably followed by 'statistical' data (when we first saw
 172    * him, how did we get his ID, how many pushes (in a timeinterval),
 173    * ...)
 174    */
 175 };
 176
 177 /***********************************************************************
 178  * /Housekeeping with peers
 179 ***********************************************************************/
 180
 181 /***********************************************************************
 182  * Globals
 183 ***********************************************************************/
 184
 185 /**
 186  * Set of all peers to keep track of them.
 187  */
 188 static struct GNUNET_CONTAINER_MultiPeerMap *peer_map;
 189
 190
 191 /**
 192  * The gossiped list of peers.
 193  */
 194 static struct GNUNET_PeerIdentity *gossip_list;
 195
 196 /**
 197  * Size of the gossiped list
 198  */
 199 //static unsigned int gossip_list_size;
 200 static uint32_t gossip_list_size;
 201
 202
 203 /**
 204  * The actual size of the sampler
 205  */
 206 static unsigned int sampler_size;
 207 //size_t sampler_size;
 208
 209 /**
 210  * The size of sampler we need to be able to satisfy the client's need of
 211  * random peers.
 212  */
 213 static unsigned int sampler_size_client_need;
 214
 215 /**
 216  * The size of sampler we need to be able to satisfy the Brahms protocol's
 217  * need of random peers.
 218  *
 219  * This is directly taken as the #gossip_list_size on update of the
 220  * #gossip_list
 221  *
 222  * This is one minimum size the sampler grows to.
 223  */
 224 static unsigned int sampler_size_est_need;
 225
 226
 227 /**
 228  * Percentage of total peer number in the gossip list
 229  * to send random PUSHes to
 230  *
 231  * TODO do not read from configuration
 232  */
 233 static float alpha;
 234
 235 /**
 236  * Percentage of total peer number in the gossip list
 237  * to send random PULLs to
 238  *
 239  * TODO do not read from configuration
 240  */
 241 static float beta;
 242
 243 /**
 244  * The percentage gamma of history updates.
 245  * Simply 1 - alpha - beta
 246  */
 247
 248
 249 /**
 250  * Identifier for the main task that runs periodically.
 251  */
 252 static struct GNUNET_SCHEDULER_Task *do_round_task;
 253
 254 /**
 255  * Time inverval the do_round task runs in.
 256  */
 257 static struct GNUNET_TIME_Relative round_interval;
 258
 259
 260
 261 /**
 262  * List to store peers received through pushes temporary.
 263  *
 264  * TODO -> multipeermap
 265  */
 266 static struct GNUNET_PeerIdentity *push_list;
 267
 268 /**
 269  * Size of the push_list;
 270  */
 271 static unsigned int push_list_size;
 272 //size_t push_list_size;
 273
 274 /**
 275  * List to store peers received through pulls temporary.
 276  *
 277  * TODO -> multipeermap
 278  */
 279 static struct GNUNET_PeerIdentity *pull_list;
 280
 281 /**
 282  * Size of the pull_list;
 283  */
 284 static unsigned int pull_list_size;
 285 //size_t pull_list_size;
 286
 287
 288 /**
 289  * Handler to NSE.
 290  */
 291 static struct GNUNET_NSE_Handle *nse;
 292
 293 /**
 294  * Handler to CADET.
 295  */
 296 static struct GNUNET_CADET_Handle *cadet_handle;
 297
 298
 299 /**
 300  * Request counter.
 301  *
 302  * Only needed in the beginning to check how many of the 64 deltas
 303  * we already have
 304  */
 305 static unsigned int req_counter;
 306
 307 /**
 308  * Time of the last request we received.
 309  *
 310  * Used to compute the expected request rate.
 311  */
 312 static struct GNUNET_TIME_Absolute last_request;
 313
 314 /**
 315  * Size of #request_deltas.
 316  */
 317 #define REQUEST_DELTAS_SIZE 64
 318 static unsigned int request_deltas_size = REQUEST_DELTAS_SIZE;
 319
 320 /**
 321  * Last 64 deltas between requests
 322  */
 323 static struct GNUNET_TIME_Relative request_deltas[REQUEST_DELTAS_SIZE];
 324
 325 /**
 326  * The prediction of the rate of requests
 327  */
 328 static struct GNUNET_TIME_Relative  request_rate;
 329
 330
 331 /**
 332  * Number of history update tasks.
 333  */
 334 uint32_t num_hist_update_tasks;
 335
 336
 337 /***********************************************************************
 338  * /Globals
 339 ***********************************************************************/
 340
 341
 342 /***********************************************************************
 343  * Util functions
 344 ***********************************************************************/
 345
 346 /**
 347  * Check if peer is already in peer array.
 348  */
 349   int
 350 in_arr (const struct GNUNET_PeerIdentity *array,
 351         unsigned int arr_size,
 352         const struct GNUNET_PeerIdentity *peer)
 353 {
 354   GNUNET_assert (NULL != peer);
 355
 356   if (0 == arr_size)
 357     return GNUNET_NO;
 358
 359   GNUNET_assert (NULL != array);
 360
 361   unsigned int i;
 362
 363   i = 0;
 364   while (0 != GNUNET_CRYPTO_cmp_peer_identity (&array[i], peer) &&
 365          i < arr_size)
 366     i++;
 367
 368   if (i == arr_size)
 369     return GNUNET_NO;
 370   else
 371     return GNUNET_YES;
 372 }
 373
 374
 375 /**
 376  * Get random peer from the gossip list.
 377  */
 378   struct GNUNET_PeerIdentity *
 379 get_rand_peer (const struct GNUNET_PeerIdentity *peer_list, unsigned int list_size)
 380 {
 381   uint32_t r_index;
 382   struct GNUNET_PeerIdentity *peer;
 383
 384   peer = GNUNET_new (struct GNUNET_PeerIdentity);
 385   // FIXME if we have only NULL in gossip list this will block
 386   // but then we might have a problem nevertheless
 387
 388   do
 389   {
 390
 391     /**;
 392      * Choose the r_index of the peer we want to return
 393      * at random from the interval of the gossip list
 394      */
 395     r_index = GNUNET_CRYPTO_random_u64 (GNUNET_CRYPTO_QUALITY_STRONG,
 396                                      list_size);
 397
 398     *peer = peer_list[r_index];
 399   } while (NULL == peer);
 400
 401   return peer;
 402 }
 403
 404
 405 /**
 406  * Get the context of a peer. If not existing, create.
 407  */
 408   struct PeerContext *
 409 get_peer_ctx (struct GNUNET_CONTAINER_MultiPeerMap *peer_map,
 410               const struct GNUNET_PeerIdentity *peer)
 411 {
 412   struct PeerContext *ctx;
 413
 414   if ( GNUNET_YES == GNUNET_CONTAINER_multipeermap_contains (peer_map, peer))
 415   {
 416     ctx = GNUNET_CONTAINER_multipeermap_get (peer_map, peer);
 417   }
 418   else
 419   {
 420     ctx = GNUNET_new (struct PeerContext);
 421     ctx->peer_flags = 0;
 422     ctx->mq = NULL;
 423     ctx->send_channel = NULL;
 424     ctx->recv_channel = NULL;
 425     ctx->outstanding_ops = NULL;
 426     ctx->num_outstanding_ops = 0;
 427     (void) GNUNET_CONTAINER_multipeermap_put (peer_map, peer, ctx,
 428                                               GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_FAST);
 429   }
 430   return ctx;
 431 }
 432
 433
 434 /**
 435  * Put random peer from sampler into the gossip list as history update.
 436  */
 437   void
 438 hist_update (void *cls, struct GNUNET_PeerIdentity *ids, uint32_t num_peers)
 439 {
 440   GNUNET_assert (1 == num_peers);
 441
 442   if (gossip_list_size < sampler_size_est_need)
 443     GNUNET_array_append (gossip_list, gossip_list_size, *ids);
 444
 445   if (0 < num_hist_update_tasks)
 446     num_hist_update_tasks--;
 447 }
 448
 449
 450 /**
 451  * Callback that is called when a channel was effectively established.
 452  * This is given to ntfy_tmt_rdy and called when the channel was
 453  * successfully established.
 454  */
 455   size_t
 456 peer_is_live (void *cls, size_t size, void *buf)
 457 {
 458   struct GNUNET_PeerIdentity *peer;
 459   struct PeerContext *peer_ctx;
 460
 461   peer = (struct GNUNET_PeerIdentity *) cls;
 462   peer_ctx = get_peer_ctx (peer_map, peer);
 463   peer_ctx->peer_flags |= LIVING;
 464
 465   LOG (GNUNET_ERROR_TYPE_DEBUG, "Peer %s is live\n", GNUNET_i2s (peer));
 466
 467   if (0 != peer_ctx->num_outstanding_ops)
 468   { /* Call outstanding operations */
 469     unsigned int i;
 470
 471     for ( i = 0 ; i < peer_ctx->num_outstanding_ops ; i++ )
 472       peer_ctx->outstanding_ops[i].op (peer_ctx->outstanding_ops[i].op_cls, peer);
 473     GNUNET_array_grow (peer_ctx->outstanding_ops, peer_ctx->num_outstanding_ops, 0);
 474   }
 475
 476   GNUNET_free (peer);
 477
 478   buf = NULL;
 479   return 0;
 480 }
 481
 482
 483 /**
 484  * Get the channel of a peer. If not existing, create.
 485  */
 486   struct GNUNET_CADET_Channel *
 487 get_channel (struct GNUNET_CONTAINER_MultiPeerMap *peer_map,
 488              const struct GNUNET_PeerIdentity *peer)
 489 {
 490   struct PeerContext *ctx;
 491   struct GNUNET_PeerIdentity *tmp_peer;
 492
 493   ctx = get_peer_ctx (peer_map, peer);
 494   if (NULL == ctx->send_channel)
 495   {
 496     ctx->send_channel = GNUNET_CADET_channel_create (cadet_handle, NULL, peer,
 497                                                      GNUNET_RPS_CADET_PORT,
 498                                                      GNUNET_CADET_OPTION_RELIABLE);
 499
 500     if (NULL == ctx->recv_channel)
 501     {
 502       tmp_peer = GNUNET_new (struct GNUNET_PeerIdentity);
 503       *tmp_peer = *peer;
 504       (void) GNUNET_CADET_notify_transmit_ready (ctx->send_channel, GNUNET_NO,
 505                                                  GNUNET_TIME_UNIT_FOREVER_REL,
 506                                                  0, peer_is_live, tmp_peer);
 507     }
 508
 509     // do I have to explicitly put it in the peer_map?
 510     (void) GNUNET_CONTAINER_multipeermap_put (peer_map, peer, ctx,
 511                                               GNUNET_CONTAINER_MULTIHASHMAPOPTION_REPLACE);
 512   }
 513   return ctx->send_channel;
 514 }
 515
 516
 517 /**
 518  * Get the message queue of a specific peer.
 519  *
 520  * If we already have a message queue open to this client,
 521  * simply return it, otherways create one.
 522  */
 523   struct GNUNET_MQ_Handle *
 524 get_mq (struct GNUNET_CONTAINER_MultiPeerMap *peer_map,
 525         const struct GNUNET_PeerIdentity *peer_id)
 526 {
 527   struct PeerContext *ctx;
 528
 529   ctx = get_peer_ctx (peer_map, peer_id);
 530   if (NULL == ctx->mq)
 531   {
 532     (void) get_channel (peer_map, peer_id);
 533     ctx->mq = GNUNET_CADET_mq_create (ctx->send_channel);
 534     //do I have to explicitly put it in the peer_map?
 535     (void) GNUNET_CONTAINER_multipeermap_put (peer_map, peer_id, ctx,
 536                                        GNUNET_CONTAINER_MULTIHASHMAPOPTION_REPLACE);
 537   }
 538   return ctx->mq;
 539 }
 540
 541
 542 /**
 543  * Sum all time relatives of an array.
 544   */
 545   struct GNUNET_TIME_Relative
 546 T_relative_sum (const struct GNUNET_TIME_Relative *rel_array, uint32_t arr_size)
 547 {
 548   struct GNUNET_TIME_Relative sum;
 549   uint32_t i;
 550
 551   sum = GNUNET_TIME_UNIT_ZERO;
 552   for ( i = 0 ; i < arr_size ; i++ )
 553   {
 554     sum = GNUNET_TIME_relative_add (sum, rel_array[i]);
 555   }
 556   return sum;
 557 }
 558
 559
 560 /**
 561  * Compute the average of given time relatives.
 562  */
 563   struct GNUNET_TIME_Relative
 564 T_relative_avg (const struct GNUNET_TIME_Relative *rel_array, uint32_t arr_size)
 565 {
 566   return GNUNET_TIME_relative_divide (T_relative_sum (rel_array, arr_size), arr_size);
 567 }
 568
 569
 570 /**
 571  * Insert PeerID in #pull_list
 572  *
 573  * Called once we know a peer is live.
 574  */
 575   void
 576 insert_in_pull_list (void *cls, const struct GNUNET_PeerIdentity *peer)
 577 {
 578   if (GNUNET_NO == in_arr (pull_list, pull_list_size, peer))
 579     GNUNET_array_append (pull_list, pull_list_size, *peer);
 580 }
 581
 582 /**
 583  * Check whether #insert_in_pull_list was already scheduled
 584  */
 585   int
 586 insert_in_pull_list_scheduled (const struct PeerContext *peer_ctx)
 587 {
 588   unsigned int i;
 589
 590   for ( i = 0 ; i < peer_ctx->num_outstanding_ops ; i++ )
 591     if (insert_in_pull_list == peer_ctx->outstanding_ops[i].op)
 592       return GNUNET_YES;
 593   return GNUNET_NO;
 594 }
 595
 596
 597 /**
 598  * Insert PeerID in #gossip_list
 599  *
 600  * Called once we know a peer is live.
 601  */
 602   void
 603 insert_in_gossip_list (void *cls, const struct GNUNET_PeerIdentity *peer)
 604 {
 605   if (GNUNET_NO == in_arr (gossip_list, gossip_list_size, peer))
 606     GNUNET_array_append (gossip_list, gossip_list_size, *peer);
 607 }
 608
 609 /**
 610  * Check whether #insert_in_pull_list was already scheduled
 611  */
 612   int
 613 insert_in_gossip_list_scheduled (const struct PeerContext *peer_ctx)
 614 {
 615   unsigned int i;
 616
 617   for ( i = 0 ; i < peer_ctx->num_outstanding_ops ; i++ )
 618     if (insert_in_gossip_list == peer_ctx->outstanding_ops[i].op)
 619       return GNUNET_YES;
 620   return GNUNET_NO;
 621 }
 622
 623
 624 /**
 625  * Update sampler with given PeerID.
 626  */
 627   void
 628 insert_in_sampler (void *cls, const struct GNUNET_PeerIdentity *peer)
 629 {
 630   RPS_sampler_update_list (peer);
 631 }
 632
 633 /**
 634  * Check whether #insert_in_sampler was already scheduled
 635  */
 636   int
 637 insert_in_sampler_scheduled (const struct PeerContext *peer_ctx)
 638 {
 639   unsigned int i;
 640
 641   for ( i = 0 ; i < peer_ctx->num_outstanding_ops ; i++ )
 642     if (insert_in_sampler== peer_ctx->outstanding_ops[i].op)
 643       return GNUNET_YES;
 644   return GNUNET_NO;
 645 }
 646
 647
 648
 649 /**
 650  * Wrapper around #RPS_sampler_resize()
 651  */
 652   void
 653 resize_wrapper ()
 654 {
 655   uint32_t bigger_size;
 656
 657   // TODO statistics
 658
 659   if (sampler_size_est_need > sampler_size_client_need)
 660     bigger_size = sampler_size_client_need;
 661   else
 662     bigger_size = sampler_size_est_need;
 663
 664   // TODO respect the request rate, min, max
 665   if (sampler_size > bigger_size*4)
 666   { /* Shrinking */
 667     RPS_sampler_resize (sampler_size/2);
 668   }
 669   else if (sampler_size < bigger_size)
 670   { /* Growing */
 671     RPS_sampler_resize (sampler_size*2);
 672   }
 673 }
 674
 675
 676 /***********************************************************************
 677  * /Util functions
 678 ***********************************************************************/
 679
 680 /**
 681  * Function called by NSE.
 682  *
 683  * Updates sizes of sampler list and gossip list and adapt those lists
 684  * accordingly.
 685  */
 686   void
 687 nse_callback (void *cls, struct GNUNET_TIME_Absolute timestamp, double logestimate, double std_dev)
 688 {
 689   double estimate;
 690   //double scale; // TODO this might go gloabal/config
 691
 692   LOG (GNUNET_ERROR_TYPE_DEBUG,
 693       "Received a ns estimate - logest: %f, std_dev: %f (old_size: %f)\n",
 694       logestimate, std_dev, sampler_size);
 695   //scale = .01;
 696   estimate = GNUNET_NSE_log_estimate_to_n (logestimate);
 697   // GNUNET_NSE_log_estimate_to_n (logestimate);
 698   estimate = pow (estimate, 1./3);
 699   // TODO add if std_dev is a number
 700   // estimate += (std_dev * scale);
 701   if ( 0 < estimate ) {
 702     LOG (GNUNET_ERROR_TYPE_DEBUG, "Changing estimate to %f\n", estimate);
 703     sampler_size_est_need = estimate;
 704   } else
 705     LOG (GNUNET_ERROR_TYPE_DEBUG, "Not using estimate %f\n", estimate);
 706
 707   /* If the NSE has changed adapt the lists accordingly */
 708   resize_wrapper ();
 709 }
 710
 711
 712 /**
 713  * Callback called once the requested PeerIDs are ready.
 714  *
 715  * Sends those to the requesting client.
 716  */
 717 void client_respond (void *cls,
 718     struct GNUNET_PeerIdentity *ids, uint32_t num_peers)
 719 {
 720   LOG (GNUNET_ERROR_TYPE_DEBUG, "sampler returned %" PRIX32 " peers\n", num_peers);
 721   struct GNUNET_MQ_Envelope *ev;
 722   struct GNUNET_RPS_CS_ReplyMessage *out_msg;
 723   struct GNUNET_SERVER_Client *client;
 724   struct client_ctx *cli_ctx;
 725
 726   client = (struct GNUNET_SERVER_Client *) cls;
 727
 728   ev = GNUNET_MQ_msg_extra (out_msg,
 729                             num_peers * sizeof (struct GNUNET_PeerIdentity),
 730                             GNUNET_MESSAGE_TYPE_RPS_CS_REPLY);
 731   out_msg->num_peers = htonl (num_peers);
 732
 733   memcpy (&out_msg[1],
 734       ids,
 735       num_peers * sizeof (struct GNUNET_PeerIdentity));
 736   GNUNET_free (ids);
 737
 738   cli_ctx = GNUNET_SERVER_client_get_user_context (client, struct client_ctx);
 739   if ( NULL == cli_ctx ) {
 740     cli_ctx = GNUNET_new (struct client_ctx);
 741     cli_ctx->mq = GNUNET_MQ_queue_for_server_client (client);
 742     GNUNET_SERVER_client_set_user_context (client, cli_ctx);
 743   }
 744
 745   GNUNET_MQ_send (cli_ctx->mq, ev);
 746 }
 747
 748
 749 /**
 750  * Handle RPS request from the client.
 751  *
 752  * @param cls closure
 753  * @param client identification of the client
 754  * @param message the actual message
 755  */
 756 static void
 757 handle_client_request (void *cls,
 758             struct GNUNET_SERVER_Client *client,
 759             const struct GNUNET_MessageHeader *message)
 760 {
 761   struct GNUNET_RPS_CS_RequestMessage *msg;
 762   uint32_t num_peers;
 763   struct GNUNET_TIME_Relative max_round_duration;
 764
 765
 766   /* Estimate request rate */
 767   if (request_deltas_size > req_counter)
 768     req_counter++;
 769   if ( 1 < req_counter)
 770   {
 771     /* Shift last request deltas to the right */
 772     memcpy (&request_deltas[1],
 773         request_deltas,
 774         (req_counter - 1) * sizeof (struct GNUNET_TIME_Relative));
 775     /* Add current delta to beginning */
 776     request_deltas[0] = GNUNET_TIME_absolute_get_difference (last_request,
 777         GNUNET_TIME_absolute_get ());
 778     request_rate = T_relative_avg (request_deltas, req_counter);
 779
 780     max_round_duration = GNUNET_TIME_relative_add (round_interval,
 781         GNUNET_TIME_relative_divide (round_interval, 2));
 782     sampler_size_client_need = max_round_duration.rel_value_us / request_rate.rel_value_us;
 783
 784     resize_wrapper ();
 785   }
 786   last_request = GNUNET_TIME_absolute_get ();
 787
 788
 789   // TODO check message size
 790   msg = (struct GNUNET_RPS_CS_RequestMessage *) message;
 791
 792   num_peers = ntohl (msg->num_peers);
 793
 794   LOG (GNUNET_ERROR_TYPE_DEBUG, "Client requested %" PRIX32 " random peer(s).\n", num_peers);
 795
 796   RPS_sampler_get_n_rand_peers (client_respond, client, num_peers, GNUNET_YES);
 797
 798   GNUNET_SERVER_receive_done (client,
 799                               GNUNET_OK);
 800 }
 801
 802
 803 /**
 804  * Handle seed from the client.
 805  *
 806  * @param cls closure
 807  * @param client identification of the client
 808  * @param message the actual message
 809  */
 810   static void
 811 handle_client_seed (void *cls,
 812             struct GNUNET_SERVER_Client *client,
 813             const struct GNUNET_MessageHeader *message)
 814 {
 815   struct GNUNET_RPS_CS_SeedMessage *in_msg;
 816   struct GNUNET_PeerIdentity *peers;
 817   uint32_t i;
 818
 819   if (sizeof (struct GNUNET_RPS_CS_SeedMessage) < ntohs (message->size))
 820   {
 821     GNUNET_break_op (0);
 822     GNUNET_SERVER_receive_done (client,
 823               GNUNET_SYSERR);
 824   }
 825   in_msg = (struct GNUNET_RPS_CS_SeedMessage *) message;
 826   if ((ntohs (message->size) - sizeof (struct GNUNET_RPS_CS_SeedMessage)) /
 827       sizeof (struct GNUNET_PeerIdentity) != ntohl (in_msg->num_peers))
 828   {
 829     GNUNET_break_op (0);
 830     GNUNET_SERVER_receive_done (client,
 831               GNUNET_SYSERR);
 832   }
 833
 834   in_msg = (struct GNUNET_RPS_CS_SeedMessage *) message;
 835   peers = (struct GNUNET_PeerIdentity *) &message[1];
 836
 837   for ( i = 0 ; i < ntohl (in_msg->num_peers) ; i++ )
 838     RPS_sampler_update_list (&peers[i]);
 839
 840   GNUNET_SERVER_receive_done (client,
 841                               GNUNET_OK);
 842 }
 843
 844
 845 /**
 846  * Handle a PUSH message from another peer.
 847  *
 848  * Check the proof of work and store the PeerID
 849  * in the temporary list for pushed PeerIDs.
 850  *
 851  * @param cls Closure
 852  * @param channel The channel the PUSH was received over
 853  * @param channel_ctx The context associated with this channel
 854  * @param msg The message header
 855  */
 856 static int
 857 handle_peer_push (void *cls,
 858     struct GNUNET_CADET_Channel *channel,
 859     void **channel_ctx,
 860     const struct GNUNET_MessageHeader *msg)
 861 {
 862   const struct GNUNET_PeerIdentity *peer;
 863
 864   // (check the proof of work)
 865
 866   peer = (const struct GNUNET_PeerIdentity *) GNUNET_CADET_channel_get_info (channel, GNUNET_CADET_OPTION_PEER);
 867   // FIXME wait for cadet to change this function
 868   LOG (GNUNET_ERROR_TYPE_DEBUG, "PUSH received (%s)\n", GNUNET_i2s (peer));
 869
 870   /* Add the sending peer to the push_list */
 871   if (GNUNET_NO == in_arr (push_list, pull_list_size, peer))
 872     GNUNET_array_append (push_list, push_list_size, *peer);
 873
 874   return GNUNET_OK;
 875 }
 876
 877 /**
 878  * Handle PULL REQUEST request message from another peer.
 879  *
 880  * Reply with the gossip list of PeerIDs.
 881  *
 882  * @param cls Closure
 883  * @param channel The channel the PUSH was received over
 884  * @param channel_ctx The context associated with this channel
 885  * @param msg The message header
 886  */
 887 static int
 888 handle_peer_pull_request (void *cls,
 889     struct GNUNET_CADET_Channel *channel,
 890     void **channel_ctx,
 891     const struct GNUNET_MessageHeader *msg)
 892 {
 893   struct GNUNET_PeerIdentity *peer;
 894   struct GNUNET_MQ_Handle *mq;
 895   struct GNUNET_MQ_Envelope *ev;
 896   struct GNUNET_RPS_P2P_PullReplyMessage *out_msg;
 897
 898
 899   peer = (struct GNUNET_PeerIdentity *) GNUNET_CADET_channel_get_info (channel,
 900                                                                        GNUNET_CADET_OPTION_PEER);
 901   // FIXME wait for cadet to change this function
 902   LOG (GNUNET_ERROR_TYPE_DEBUG,
 903       "PULL REQUEST from peer %s received, going to send %u peers\n",
 904       GNUNET_i2s (peer), gossip_list_size);
 905
 906   mq = get_mq (peer_map, peer);
 907
 908   ev = GNUNET_MQ_msg_extra (out_msg,
 909                            gossip_list_size * sizeof (struct GNUNET_PeerIdentity),
 910                            GNUNET_MESSAGE_TYPE_RPS_PP_PULL_REPLY);
 911   //out_msg->num_peers = htonl (gossip_list_size);
 912   out_msg->num_peers = htonl (gossip_list_size);
 913   memcpy (&out_msg[1], gossip_list,
 914          gossip_list_size * sizeof (struct GNUNET_PeerIdentity));
 915
 916   GNUNET_MQ_send (mq, ev);
 917
 918   return GNUNET_OK;
 919 }
 920
 921 /**
 922  * Handle PULL REPLY message from another peer.
 923  *
 924  * Check whether we sent a corresponding request and
 925  * whether this reply is the first one.
 926  *
 927  * @param cls Closure
 928  * @param channel The channel the PUSH was received over
 929  * @param channel_ctx The context associated with this channel
 930  * @param msg The message header
 931  */
 932   static int
 933 handle_peer_pull_reply (void *cls,
 934     struct GNUNET_CADET_Channel *channel,
 935     void **channel_ctx,
 936     const struct GNUNET_MessageHeader *msg)
 937 {
 938   LOG (GNUNET_ERROR_TYPE_DEBUG, "PULL REPLY received\n");
 939
 940   struct GNUNET_RPS_P2P_PullReplyMessage *in_msg;
 941   struct GNUNET_PeerIdentity *peers;
 942   struct PeerContext *peer_ctx;
 943   struct PeerOutstandingOp out_op;
 944   uint32_t i;
 945
 946   if (sizeof (struct GNUNET_RPS_P2P_PullReplyMessage) > ntohs (msg->size))
 947   {
 948     GNUNET_break_op (0); // At the moment our own implementation seems to break that.
 949     return GNUNET_SYSERR;
 950   }
 951   in_msg = (struct GNUNET_RPS_P2P_PullReplyMessage *) msg;
 952   if ((ntohs (msg->size) - sizeof (struct GNUNET_RPS_P2P_PullReplyMessage)) / sizeof (struct GNUNET_PeerIdentity) != ntohl (in_msg->num_peers))
 953   {
 954     LOG (GNUNET_ERROR_TYPE_ERROR, "message says it sends %" PRIu64 " peers, have space for %i peers\n",
 955         ntohl (in_msg->num_peers),
 956         (ntohs (msg->size) - sizeof (struct GNUNET_RPS_P2P_PullReplyMessage)) / sizeof (struct GNUNET_PeerIdentity));
 957     GNUNET_break_op (0);
 958     return GNUNET_SYSERR;
 959   }
 960
 961   // TODO check that we sent a request and that it is the first reply
 962
 963   peers = (struct GNUNET_PeerIdentity *) &msg[1];
 964   for ( i = 0 ; i < ntohl (in_msg->num_peers) ; i++ )
 965   {
 966     peer_ctx = get_peer_ctx (peer_map, &peers[i]);
 967
 968     if ((0 != (peer_ctx->peer_flags && LIVING)) ||
 969         NULL != peer_ctx->recv_channel)
 970     {
 971       if (GNUNET_NO == in_arr (pull_list, pull_list_size, &peers[i]))
 972         GNUNET_array_append (pull_list, pull_list_size, peers[i]);
 973     }
 974     else if (GNUNET_NO == insert_in_pull_list_scheduled (peer_ctx))
 975     {
 976       out_op.op = insert_in_pull_list;
 977       GNUNET_array_append (peer_ctx->outstanding_ops, peer_ctx->num_outstanding_ops, out_op);
 978     }
 979   }
 980
 981   // TODO check that id is valid - whether it is reachable
 982
 983   return GNUNET_OK;
 984 }
 985
 986
 987 /**
 988  * Send out PUSHes and PULLs.
 989  *
 990  * This is executed regylary.
 991  */
 992 static void
 993 do_round (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
 994 {
 995   LOG (GNUNET_ERROR_TYPE_DEBUG, "Going to execute next round\n");
 996
 997   uint32_t i;
 998   //unsigned int *n_arr;
 999   unsigned int n_peers; /* Number of peers we send pushes/pulls to */
1000   struct GNUNET_MQ_Envelope *ev;
1001   const struct GNUNET_PeerIdentity *peer;
1002   struct GNUNET_MQ_Handle *mq;
1003
1004   // TODO log lists, ...
1005
1006
1007   /* Would it make sense to have one shuffeled gossip list and then
1008    * to send PUSHes to first alpha peers, PULL requests to next beta peers and
1009    * use the rest to update sampler?
1010    * in essence get random peers with consumption */
1011
1012   /* Send PUSHes */
1013   //n_arr = GNUNET_CRYPTO_random_permute (GNUNET_CRYPTO_QUALITY_STRONG, (unsigned int) gossip_list_size);
1014   if (0 != gossip_list_size)
1015   {
1016     n_peers = round (alpha * gossip_list_size);
1017     if (0 == n_peers)
1018       n_peers = 1;
1019     LOG (GNUNET_ERROR_TYPE_DEBUG, "Going to send pushes to %u (%f * %u) peers.\n",
1020         n_peers, alpha, gossip_list_size);
1021     for ( i = 0 ; i < n_peers ; i++ )
1022     {
1023       peer = get_rand_peer (gossip_list, gossip_list_size);
1024       if (own_identity != peer)
1025       { // FIXME if this fails schedule/loop this for later
1026         LOG (GNUNET_ERROR_TYPE_DEBUG, "Sending PUSH to peer %s of gossiped list.\n", GNUNET_i2s (peer));
1027
1028         ev = GNUNET_MQ_msg_header (GNUNET_MESSAGE_TYPE_RPS_PP_PUSH);
1029         // FIXME sometimes it returns a pointer to a freed mq
1030         mq = get_mq (peer_map, peer);
1031         GNUNET_MQ_send (mq, ev);
1032       }
1033     }
1034   }
1035
1036
1037   /* Send PULL requests */
1038   //n_arr = GNUNET_CRYPTO_random_permute (GNUNET_CRYPTO_QUALITY_STRONG, (unsigned int) sampler_list->size);
1039   if (0 != gossip_list_size)
1040   {
1041     n_peers = round (beta * gossip_list_size);
1042     if (0 == n_peers)
1043       n_peers = 1;
1044     LOG (GNUNET_ERROR_TYPE_DEBUG, "Going to send pulls to %u (%f * %u) peers.\n",
1045         n_peers, beta, gossip_list_size);
1046     for ( i = 0 ; i < n_peers ; i++ )
1047     {
1048       peer = get_rand_peer (gossip_list, gossip_list_size);
1049       if (own_identity != peer)
1050       { // FIXME if this fails schedule/loop this for later
1051         LOG (GNUNET_ERROR_TYPE_DEBUG, "Sending PULL request to peer %s of gossiped list.\n", GNUNET_i2s (peer));
1052
1053         ev = GNUNET_MQ_msg_header (GNUNET_MESSAGE_TYPE_RPS_PP_PULL_REQUEST);
1054         //pull_msg = NULL;
1055         mq = get_mq (peer_map, peer);
1056         GNUNET_MQ_send (mq, ev);
1057       }
1058     }
1059   }
1060
1061
1062   /* Update gossip list */
1063   uint32_t r_index;
1064
1065   if ( push_list_size <= alpha * gossip_list_size &&
1066        push_list_size != 0 &&
1067        pull_list_size != 0 )
1068   {
1069     LOG (GNUNET_ERROR_TYPE_DEBUG, "Update of the gossip list.\n");
1070
1071     uint32_t first_border;
1072     uint32_t second_border;
1073
1074     first_border = round (alpha * sampler_size_est_need);
1075     second_border = first_border + round (beta * sampler_size_est_need);
1076
1077     GNUNET_array_grow (gossip_list, gossip_list_size, second_border);
1078
1079     for ( i = 0 ; i < first_border ; i++ )
1080     { // TODO use RPS_sampler_get_n_rand_peers
1081       /* Update gossip list with peers received through PUSHes */
1082       r_index = GNUNET_CRYPTO_random_u64 (GNUNET_CRYPTO_QUALITY_STRONG,
1083                                        push_list_size);
1084       gossip_list[i] = push_list[r_index];
1085       // TODO change the peer_flags accordingly
1086     }
1087
1088     for ( i = first_border ; i < second_border ; i++ )
1089     {
1090       /* Update gossip list with peers received through PULLs */
1091       r_index = GNUNET_CRYPTO_random_u64 (GNUNET_CRYPTO_QUALITY_STRONG,
1092                                        pull_list_size);
1093       gossip_list[i] = pull_list[r_index];
1094       // TODO change the peer_flags accordingly
1095     }
1096
1097     for ( i = second_border ; i < gossip_list_size ; i++ )
1098     {
1099       /* Update gossip list with peers from history */
1100       RPS_sampler_get_n_rand_peers (hist_update, NULL, 1, GNUNET_NO);
1101       num_hist_update_tasks++;
1102       // TODO change the peer_flags accordingly
1103     }
1104
1105   }
1106   else
1107   {
1108     LOG (GNUNET_ERROR_TYPE_DEBUG, "No update of the gossip list. ()\n");
1109   }
1110   // TODO independent of that also get some peers from CADET_get_peers()?
1111
1112
1113   /* Update samplers */
1114
1115   for ( i = 0 ; i < push_list_size ; i++ )
1116   {
1117     RPS_sampler_update_list (&push_list[i]);
1118     // TODO set in_flag?
1119   }
1120
1121   for ( i = 0 ; i < pull_list_size ; i++ )
1122   {
1123     RPS_sampler_update_list (&pull_list[i]);
1124     // TODO set in_flag?
1125   }
1126
1127
1128   /* Empty push/pull lists */
1129   GNUNET_array_grow (push_list, push_list_size, 0);
1130   GNUNET_array_grow (pull_list, pull_list_size, 0);
1131
1132   struct GNUNET_TIME_Relative time_next_round;
1133   struct GNUNET_TIME_Relative half_round_interval;
1134   unsigned int rand_delay;
1135
1136   /* Compute random time value between .5 * round_interval and 1.5 *round_interval */
1137   half_round_interval = GNUNET_TIME_relative_divide (round_interval, 2);
1138   do
1139   {
1140   /*
1141    * Compute random value between (0 and 1) * round_interval
1142    * via multiplying round_interval with a 'fraction' (0 to value)/value
1143    */
1144   rand_delay = GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK, UINT_MAX/10);
1145   time_next_round = GNUNET_TIME_relative_multiply (round_interval,  rand_delay);
1146   time_next_round = GNUNET_TIME_relative_divide   (time_next_round, UINT_MAX/10);
1147   time_next_round = GNUNET_TIME_relative_add      (time_next_round, half_round_interval);
1148   } while (GNUNET_TIME_UNIT_FOREVER_REL.rel_value_us == time_next_round.rel_value_us);
1149
1150   /* Schedule next round */
1151   do_round_task = GNUNET_SCHEDULER_add_delayed (round_interval, &do_round, NULL);
1152   LOG (GNUNET_ERROR_TYPE_DEBUG, "Finished round\n");
1153 }
1154
1155
1156 /**
1157  * Open a connection to given peer and store channel and mq.
1158  */
1159   void
1160 insertCB (void *cls, const struct GNUNET_PeerIdentity *id)
1161 {
1162   // We open a channel to be notified when this peer goes down.
1163   (void) get_channel (peer_map, id);
1164 }
1165
1166
1167 /**
1168  * Close the connection to given peer and delete channel and mq.
1169  */
1170   void
1171 removeCB (void *cls, const struct GNUNET_PeerIdentity *id)
1172 {
1173   size_t s;
1174   struct PeerContext *ctx;
1175
1176   s = RPS_sampler_count_id (id);
1177   if ( 1 >= s )
1178   {
1179     if (GNUNET_YES == GNUNET_CONTAINER_multipeermap_contains (peer_map, id))
1180     {
1181       ctx = GNUNET_CONTAINER_multipeermap_get (peer_map, id);
1182       if (NULL != ctx->send_channel)
1183       {
1184         if (NULL != ctx->mq)
1185         {
1186           GNUNET_MQ_destroy (ctx->mq);
1187         }
1188         // may already be freed at shutdown of cadet
1189         //GNUNET_CADET_channel_destroy (ctx->send_channel);
1190       }
1191       // TODO cleanup peer
1192       (void) GNUNET_CONTAINER_multipeermap_remove_all (peer_map, id);
1193     }
1194   }
1195 }
1196
1197 static void
1198 rps_start (struct GNUNET_SERVER_Handle *server);
1199
1200 /**
1201  * This is called from GNUNET_CADET_get_peers().
1202  *
1203  * It is called on every peer(ID) that cadet somehow has contact with.
1204  * We use those to initialise the sampler.
1205  */
1206 void
1207 init_peer_cb (void *cls,
1208               const struct GNUNET_PeerIdentity *peer,
1209               int tunnel, // "Do we have a tunnel towards this peer?"
1210               unsigned int n_paths, // "Number of known paths towards this peer"
1211               unsigned int best_path) // "How long is the best path?
1212                                       // (0 = unknown, 1 = ourselves, 2 = neighbor)"
1213 {
1214   struct GNUNET_SERVER_Handle *server;
1215   struct PeerOutstandingOp out_op;
1216   struct PeerContext *peer_ctx;
1217
1218   server = (struct GNUNET_SERVER_Handle *) cls;
1219   if ( NULL != peer )
1220   {
1221     LOG (GNUNET_ERROR_TYPE_DEBUG,
1222         "Got peer %s (at %p) from CADET (gossip_list_size: %u)\n",
1223         GNUNET_i2s (peer), peer, gossip_list_size);
1224
1225     // maybe create a function for that
1226     peer_ctx = get_peer_ctx (peer_map, peer);
1227     if (GNUNET_NO == insert_in_sampler_scheduled (peer_ctx))
1228     {
1229       out_op.op = insert_in_sampler;
1230       GNUNET_array_append (peer_ctx->outstanding_ops, peer_ctx->num_outstanding_ops, out_op);
1231     }
1232
1233     if (GNUNET_NO == insert_in_gossip_list_scheduled (peer_ctx))
1234     {
1235       out_op.op = insert_in_gossip_list;
1236       GNUNET_array_append (peer_ctx->outstanding_ops, peer_ctx->num_outstanding_ops, out_op);
1237     }
1238
1239     /* Issue livelyness test on peer */
1240     (void) get_channel (peer_map, peer);
1241
1242     // send push/pull to each of those peers?
1243   }
1244   else
1245     rps_start (server);
1246 }
1247
1248
1249 /**
1250  * Callback used to clean the multipeermap.
1251  */
1252   int
1253 peer_remove_cb (void *cls, const struct GNUNET_PeerIdentity *key, void *value)
1254 {
1255   struct PeerContext *peer_ctx;
1256
1257   peer_ctx = (struct PeerContext *) value;
1258
1259   if ( NULL != peer_ctx->mq)
1260     GNUNET_MQ_destroy (peer_ctx->mq);
1261
1262   if ( NULL != peer_ctx->send_channel)
1263     GNUNET_CADET_channel_destroy (peer_ctx->send_channel);
1264
1265   if ( NULL != peer_ctx->recv_channel)
1266     GNUNET_CADET_channel_destroy (peer_ctx->recv_channel);
1267
1268   if (GNUNET_NO == GNUNET_CONTAINER_multipeermap_remove_all (peer_map, key))
1269     LOG (GNUNET_ERROR_TYPE_WARNING, "removing peer from peer_map failed\n");
1270
1271   return GNUNET_YES;
1272 }
1273
1274
1275 /**
1276  * Task run during shutdown.
1277  *
1278  * @param cls unused
1279  * @param tc unused
1280  */
1281 static void
1282 shutdown_task (void *cls,
1283                const struct GNUNET_SCHEDULER_TaskContext *tc)
1284 {
1285   LOG (GNUNET_ERROR_TYPE_DEBUG, "RPS is going down\n");
1286
1287   if ( NULL != do_round_task )
1288   {
1289     GNUNET_SCHEDULER_cancel (do_round_task);
1290     do_round_task = NULL;
1291   }
1292
1293
1294   if (GNUNET_SYSERR == GNUNET_CONTAINER_multipeermap_iterate (peer_map, peer_remove_cb, NULL))
1295     LOG (GNUNET_ERROR_TYPE_WARNING,
1296         "Iterating over peers to disconnect from them was cancelled\n");
1297
1298   GNUNET_CONTAINER_multipeermap_destroy (peer_map);
1299
1300   GNUNET_NSE_disconnect (nse);
1301   GNUNET_CADET_disconnect (cadet_handle);
1302   GNUNET_free (own_identity);
1303   RPS_sampler_destroy ();
1304   GNUNET_array_grow (request_deltas, request_deltas_size, 0);
1305   GNUNET_array_grow (gossip_list, gossip_list_size, 0);
1306   GNUNET_array_grow (push_list, push_list_size, 0);
1307   GNUNET_array_grow (pull_list, pull_list_size, 0);
1308 }
1309
1310
1311 /**
1312  * A client disconnected.  Remove all of its data structure entries.
1313  *
1314  * @param cls closure, NULL
1315  * @param client identification of the client
1316  */
1317 static void
1318 handle_client_disconnect (void *cls,
1319                           struct GNUNET_SERVER_Client * client)
1320 {
1321 }
1322
1323 /**
1324  * Handle the channel a peer opens to us.
1325  *
1326  * @param cls The closure
1327  * @param channel The channel the peer wants to establish
1328  * @param initiator The peer's peer ID
1329  * @param port The port the channel is being established over
1330  * @param options Further options
1331  */
1332   static void *
1333 handle_inbound_channel (void *cls,
1334                         struct GNUNET_CADET_Channel *channel,
1335                         const struct GNUNET_PeerIdentity *initiator,
1336                         uint32_t port,
1337                         enum GNUNET_CADET_ChannelOption options)
1338 {
1339   struct PeerContext *ctx;
1340
1341   LOG (GNUNET_ERROR_TYPE_DEBUG,
1342       "New channel was established to us (Peer %s).\n",
1343       GNUNET_i2s (initiator));
1344
1345   GNUNET_assert (NULL != channel);
1346
1347   // we might not even store the recv_channel
1348
1349   ctx = get_peer_ctx (peer_map, initiator);
1350   if (NULL != ctx->recv_channel)
1351   {
1352     ctx->recv_channel = channel;
1353   }
1354
1355   // FIXME there might already be an established channel
1356
1357   //ctx->peer_flags = IN_OTHER_GOSSIP_LIST;
1358   ctx->mq = NULL; // TODO create mq?
1359
1360   (void) GNUNET_CONTAINER_multipeermap_put (peer_map, initiator, ctx,
1361       GNUNET_CONTAINER_MULTIHASHMAPOPTION_REPLACE);
1362   return NULL; // TODO
1363 }
1364
1365 /**
1366  * This is called when a remote peer destroys a channel.
1367  *
1368  * @param cls The closure
1369  * @param channel The channel being closed
1370  * @param channel_ctx The context associated with this channel
1371  */
1372   static void
1373 cleanup_channel (void *cls,
1374                 const struct GNUNET_CADET_Channel *channel,
1375                 void *channel_ctx)
1376 {
1377   struct GNUNET_PeerIdentity *peer;
1378   struct PeerContext *peer_ctx;
1379
1380   LOG (GNUNET_ERROR_TYPE_DEBUG, "Channel to remote peer was destroyed.\n");
1381
1382   peer = (struct GNUNET_PeerIdentity *) GNUNET_CADET_channel_get_info (
1383       (struct GNUNET_CADET_Channel *) channel, GNUNET_CADET_OPTION_PEER);
1384        // Guess simply casting isn't the nicest way...
1385        // FIXME wait for cadet to change this function
1386   RPS_sampler_reinitialise_by_value (peer);
1387
1388   peer_ctx = GNUNET_CONTAINER_multipeermap_get (peer_map, peer);
1389   /* Somwewhat {ab,re}use the iterator function */
1390   (void) peer_remove_cb (peer, peer, peer_ctx);
1391 }
1392
1393
1394 /**
1395  * Actually start the service.
1396  */
1397   static void
1398 rps_start (struct GNUNET_SERVER_Handle *server)
1399 {
1400   static const struct GNUNET_SERVER_MessageHandler handlers[] = {
1401     {&handle_client_request, NULL, GNUNET_MESSAGE_TYPE_RPS_CS_REQUEST,
1402       sizeof (struct GNUNET_RPS_CS_RequestMessage)},
1403     {&handle_client_seed,    NULL, GNUNET_MESSAGE_TYPE_RPS_CS_SEED, 0},
1404     {NULL, NULL, 0, 0}
1405   };
1406
1407   GNUNET_SERVER_add_handlers (server, handlers);
1408   GNUNET_SERVER_disconnect_notify (server,
1409                                    &handle_client_disconnect,
1410                                    NULL);
1411   LOG (GNUNET_ERROR_TYPE_DEBUG, "Ready to receive requests from clients\n");
1412
1413
1414   num_hist_update_tasks = 0;
1415
1416   do_round_task = GNUNET_SCHEDULER_add_now (&do_round, NULL);
1417   LOG (GNUNET_ERROR_TYPE_DEBUG, "Scheduled first round\n");
1418
1419   GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_FOREVER_REL,
1420                                 &shutdown_task,
1421                                 NULL);
1422 }
1423
1424
1425 /**
1426  * Process statistics requests.
1427  *
1428  * @param cls closure
1429  * @param server the initialized server
1430  * @param c configuration to use
1431  */
1432   static void
1433 run (void *cls,
1434      struct GNUNET_SERVER_Handle *server,
1435      const struct GNUNET_CONFIGURATION_Handle *c)
1436 {
1437   // TODO check what this does -- copied from gnunet-boss
1438   // - seems to work as expected
1439   GNUNET_log_setup ("rps", GNUNET_error_type_to_string (GNUNET_ERROR_TYPE_DEBUG), NULL);
1440
1441   LOG (GNUNET_ERROR_TYPE_DEBUG, "RPS started\n");
1442
1443
1444   cfg = c;
1445
1446
1447   /* Get own ID */
1448   own_identity = GNUNET_new (struct GNUNET_PeerIdentity);
1449   GNUNET_CRYPTO_get_peer_identity (cfg, own_identity); // TODO check return value
1450   GNUNET_assert (NULL != own_identity);
1451   LOG (GNUNET_ERROR_TYPE_DEBUG, "Own identity is %s (at %p).\n", GNUNET_i2s (own_identity), own_identity);
1452
1453
1454   /* Get time interval from the configuration */
1455   if (GNUNET_OK != GNUNET_CONFIGURATION_get_value_time (cfg, "RPS",
1456                                                         "ROUNDINTERVAL",
1457                                                         &round_interval))
1458   {
1459     LOG (GNUNET_ERROR_TYPE_DEBUG, "Failed to read ROUNDINTERVAL from config\n");
1460     GNUNET_SCHEDULER_shutdown ();
1461     return;
1462   }
1463
1464   /* Get initial size of sampler/gossip list from the configuration */
1465   if (GNUNET_OK != GNUNET_CONFIGURATION_get_value_number (cfg, "RPS",
1466                                                          "INITSIZE",
1467                                                          (long long unsigned int *) &sampler_size_est_need))
1468   {
1469     LOG (GNUNET_ERROR_TYPE_DEBUG, "Failed to read INITSIZE from config\n");
1470     GNUNET_SCHEDULER_shutdown ();
1471     return;
1472   }
1473   LOG (GNUNET_ERROR_TYPE_DEBUG, "INITSIZE is %" PRIu64 "\n", sampler_size_est_need);
1474
1475
1476   gossip_list = NULL;
1477
1478
1479   /* connect to NSE */
1480   nse = GNUNET_NSE_connect (cfg, nse_callback, NULL);
1481   // TODO check whether that was successful
1482   LOG (GNUNET_ERROR_TYPE_DEBUG, "Connected to NSE\n");
1483
1484
1485   alpha = 0.45;
1486   beta  = 0.45;
1487   // TODO initialise thresholds - ?
1488
1489   /* Get alpha from the configuration */
1490   if (GNUNET_OK != GNUNET_CONFIGURATION_get_value_float (cfg, "RPS",
1491                                                          "ALPHA",
1492                                                          &alpha))
1493   {
1494     LOG (GNUNET_ERROR_TYPE_DEBUG, "No ALPHA specified in the config\n");
1495   }
1496   LOG (GNUNET_ERROR_TYPE_DEBUG, "ALPHA is %f\n", alpha);
1497
1498   /* Get beta from the configuration */
1499   if (GNUNET_OK != GNUNET_CONFIGURATION_get_value_float (cfg, "RPS",
1500                                                          "BETA",
1501                                                          &beta))
1502   {
1503     LOG (GNUNET_ERROR_TYPE_DEBUG, "No BETA specified in the config\n");
1504   }
1505   LOG (GNUNET_ERROR_TYPE_DEBUG, "BETA is %f\n", beta);
1506
1507   // TODO check that alpha + beta < 1
1508
1509   peer_map = GNUNET_CONTAINER_multipeermap_create (sampler_size_est_need, GNUNET_NO);
1510
1511
1512   /* Initialise cadet */
1513   static const struct GNUNET_CADET_MessageHandler cadet_handlers[] = {
1514     {&handle_peer_push        , GNUNET_MESSAGE_TYPE_RPS_PP_PUSH        ,
1515       sizeof (struct GNUNET_MessageHeader)},
1516     {&handle_peer_pull_request, GNUNET_MESSAGE_TYPE_RPS_PP_PULL_REQUEST,
1517       sizeof (struct GNUNET_MessageHeader)},
1518     {&handle_peer_pull_reply  , GNUNET_MESSAGE_TYPE_RPS_PP_PULL_REPLY  , 0},
1519     {NULL, 0, 0}
1520   };
1521
1522   const uint32_t ports[] = {GNUNET_RPS_CADET_PORT, 0}; // _PORT specified in src/rps/rps.h
1523   cadet_handle = GNUNET_CADET_connect (cfg,
1524                                        cls,
1525                                        &handle_inbound_channel,
1526                                        &cleanup_channel,
1527                                        cadet_handlers,
1528                                        ports);
1529   LOG (GNUNET_ERROR_TYPE_DEBUG, "Connected to CADET\n");
1530
1531
1532   /* Initialise sampler */
1533   struct GNUNET_TIME_Relative half_round_interval;
1534   struct GNUNET_TIME_Relative  max_round_interval;
1535
1536   half_round_interval = GNUNET_TIME_relative_multiply (round_interval, .5);
1537   max_round_interval = GNUNET_TIME_relative_add (round_interval, half_round_interval);
1538
1539   RPS_sampler_init (sampler_size_est_need, max_round_interval,
1540       insertCB, NULL, removeCB, NULL);
1541   sampler_size = sampler_size_est_need;
1542
1543   /* Initialise push and pull maps */
1544   push_list = NULL;
1545   push_list_size = 0;
1546   pull_list = NULL;
1547   pull_list_size = 0;
1548
1549
1550   LOG (GNUNET_ERROR_TYPE_DEBUG, "Requesting peers from CADET\n");
1551   GNUNET_CADET_get_peers (cadet_handle, &init_peer_cb, server);
1552
1553   // TODO send push/pull to each of those peers?
1554 }
1555
1556
1557 /**
1558  * The main function for the rps service.
1559  *
1560  * @param argc number of arguments from the command line
1561  * @param argv command line arguments
1562  * @return 0 ok, 1 on error
1563  */
1564   int
1565 main (int argc, char *const *argv)
1566 {
1567   return (GNUNET_OK ==
1568           GNUNET_SERVICE_run (argc,
1569                               argv,
1570                               "rps",
1571                               GNUNET_SERVICE_OPTION_NONE,
1572                               &run, NULL)) ? 0 : 1;
1573 }
1574
1575 /* end of gnunet-service-rps.c */