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