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