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