src/rps/test_rps.c

   1 /*
   2      This file is part of GNUnet.
   3      Copyright (C) 2009, 2012 GNUnet e.V.
   4
   5      GNUnet is free software: you can redistribute it and/or modify it
   6      under the terms of the GNU Affero General Public License as published
   7      by the Free Software Foundation, either version 3 of the License,
   8      or (at your 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      Affero General Public License for more details.
  14
  15      You should have received a copy of the GNU Affero General Public License
  16      along with this program.  If not, see <http://www.gnu.org/licenses/>.
  17
  18      SPDX-License-Identifier: AGPL3.0-or-later
  19 */
  20 /**
  21  * @file rps/test_rps.c
  22  * @brief Testcase for the random peer sampling service.  Starts
  23  *        a peergroup with a given number of peers, then waits to
  24  *        receive size pushes/pulls from each peer.  Expects to wait
  25  *        for one message from each peer.
  26  */
  27 #include "platform.h"
  28 #include "gnunet_util_lib.h"
  29 #include "gnunet_testbed_service.h"
  30
  31 #include "gnunet_rps_service.h"
  32 #include "rps-test_util.h"
  33 #include "gnunet-service-rps_sampler_elem.h"
  34
  35 #include <inttypes.h>
  36
  37
  38 /**
  39  * How many peers do we start?
  40  */
  41 static uint32_t num_peers;
  42
  43 /**
  44  * How long do we run the test?
  45  * In seconds.
  46  */
  47 static uint32_t timeout_s;
  48
  49 /**
  50  * How long do we run the test?
  51  */
  52 //#define TIMEOUT GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 30)
  53 static struct GNUNET_TIME_Relative timeout;
  54
  55
  56 /**
  57  * Portion of malicious peers
  58  */
  59 static double portion = .1;
  60
  61 /**
  62  * Type of malicious peer to test
  63  */
  64 static unsigned int mal_type = 0;
  65
  66 /**
  67  * Handles to all of the running peers
  68  */
  69 static struct GNUNET_TESTBED_Peer **testbed_peers;
  70
  71 /**
  72  * @brief Indicates whether peer should go off- or online
  73  */
  74 enum PEER_ONLINE_DELTA {
  75   /**
  76    * @brief Indicates peer going online
  77    */
  78   PEER_GO_ONLINE = 1,
  79   /**
  80    * @brief Indicates peer going offline
  81    */
  82   PEER_GO_OFFLINE = -1,
  83 };
  84
  85 /**
  86  * Operation map entry
  87  */
  88 struct OpListEntry
  89 {
  90   /**
  91    * DLL next ptr
  92    */
  93   struct OpListEntry *next;
  94
  95   /**
  96    * DLL prev ptr
  97    */
  98   struct OpListEntry *prev;
  99
 100   /**
 101    * The testbed operation
 102    */
 103   struct GNUNET_TESTBED_Operation *op;
 104
 105   /**
 106    * Depending on whether we start or stop RPS service at the peer, set this to
 107    * #PEER_GO_ONLINE (1) or #PEER_GO_OFFLINE (-1)
 108    */
 109   enum PEER_ONLINE_DELTA delta;
 110
 111   /**
 112    * Index of the regarding peer
 113    */
 114   unsigned int index;
 115 };
 116
 117 /**
 118  * OpList DLL head
 119  */
 120 static struct OpListEntry *oplist_head;
 121
 122 /**
 123  * OpList DLL tail
 124  */
 125 static struct OpListEntry *oplist_tail;
 126
 127
 128 /**
 129  * A pending reply: A request was sent and the reply is pending.
 130  */
 131 struct PendingReply
 132 {
 133   /**
 134    * DLL next,prev ptr
 135    */
 136   struct PendingReply *next;
 137   struct PendingReply *prev;
 138
 139   /**
 140    * Handle to the request we are waiting for
 141    */
 142   struct GNUNET_RPS_Request_Handle *req_handle;
 143
 144   /**
 145    * The peer that requested
 146    */
 147   struct RPSPeer *rps_peer;
 148 };
 149
 150
 151 /**
 152  * A pending request: A request was not made yet but is scheduled for later.
 153  */
 154 struct PendingRequest
 155 {
 156   /**
 157    * DLL next,prev ptr
 158    */
 159   struct PendingRequest *next;
 160   struct PendingRequest *prev;
 161
 162   /**
 163    * Handle to the request we are waiting for
 164    */
 165   struct GNUNET_SCHEDULER_Task *request_task;
 166
 167   /**
 168    * The peer that requested
 169    */
 170   struct RPSPeer *rps_peer;
 171 };
 172
 173
 174 /**
 175  * Information we track for each peer.
 176  */
 177 struct RPSPeer
 178 {
 179   /**
 180    * Index of the peer.
 181    */
 182   unsigned int index;
 183
 184   /**
 185    * Handle for RPS connect operation.
 186    */
 187   struct GNUNET_TESTBED_Operation *op;
 188
 189   /**
 190    * Handle to RPS service.
 191    */
 192   struct GNUNET_RPS_Handle *rps_handle;
 193
 194   /**
 195    * Handle to stream requests
 196    */
 197   struct GNUNET_RPS_StreamRequestHandle *rps_srh;
 198
 199   /**
 200    * ID of the peer.
 201    */
 202   struct GNUNET_PeerIdentity *peer_id;
 203
 204   /**
 205    * A request handle to check for an request
 206    */
 207   //struct GNUNET_RPS_Request_Handle *req_handle;
 208
 209   /**
 210    * Peer on- or offline?
 211    */
 212   int online;
 213
 214   /**
 215    * Number of Peer IDs to request during the whole test
 216    */
 217   unsigned int num_ids_to_request;
 218
 219   /**
 220    * Pending requests DLL
 221    */
 222   struct PendingRequest *pending_req_head;
 223   struct PendingRequest *pending_req_tail;
 224
 225   /**
 226    * Number of pending requests
 227    */
 228   unsigned int num_pending_reqs;
 229
 230   /**
 231    * Pending replies DLL
 232    */
 233   struct PendingReply *pending_rep_head;
 234   struct PendingReply *pending_rep_tail;
 235
 236   /**
 237    * Number of pending replies
 238    */
 239   unsigned int num_pending_reps;
 240
 241   /**
 242    * Number of received PeerIDs
 243    */
 244   unsigned int num_recv_ids;
 245
 246   /**
 247    * Pending operation on that peer
 248    */
 249   const struct OpListEntry *entry_op_manage;
 250
 251   /**
 252    * Testbed operation to connect to statistics service
 253    */
 254   struct GNUNET_TESTBED_Operation *stat_op;
 255
 256   /**
 257    * Handle to the statistics service
 258    */
 259   struct GNUNET_STATISTICS_Handle *stats_h;
 260
 261   /**
 262    * @brief flags to indicate which statistics values have been already
 263    * collected from the statistics service.
 264    * Used to check whether we are able to shutdown.
 265    */
 266   uint32_t stat_collected_flags;
 267
 268   /**
 269    * @brief File name of the file the stats are finally written to
 270    */
 271   const char *file_name_stats;
 272
 273   /**
 274    * @brief File name of the file the stats are finally written to
 275    */
 276   const char *file_name_probs;
 277
 278   /**
 279    * @brief The current view
 280    */
 281   struct GNUNET_PeerIdentity *cur_view;
 282
 283   /**
 284    * @brief Number of peers in the #cur_view.
 285    */
 286   uint32_t cur_view_count;
 287
 288   /**
 289    * @brief Number of occurrences in other peer's view
 290    */
 291   uint32_t count_in_views;
 292
 293   /**
 294    * @brief statistics values
 295    */
 296   uint64_t num_rounds;
 297   uint64_t num_blocks;
 298   uint64_t num_blocks_many_push;
 299   uint64_t num_blocks_no_push;
 300   uint64_t num_blocks_no_pull;
 301   uint64_t num_blocks_many_push_no_pull;
 302   uint64_t num_blocks_no_push_no_pull;
 303   uint64_t num_issued_push;
 304   uint64_t num_issued_pull_req;
 305   uint64_t num_issued_pull_rep;
 306   uint64_t num_sent_push;
 307   uint64_t num_sent_pull_req;
 308   uint64_t num_sent_pull_rep;
 309   uint64_t num_recv_push;
 310   uint64_t num_recv_pull_req;
 311   uint64_t num_recv_pull_rep;
 312 };
 313
 314 enum STAT_TYPE
 315 {
 316   STAT_TYPE_ROUNDS                    =    0x1, /*   1 */
 317   STAT_TYPE_BLOCKS                    =    0x2, /*   2 */
 318   STAT_TYPE_BLOCKS_MANY_PUSH          =    0x4, /*   3 */
 319   STAT_TYPE_BLOCKS_NO_PUSH            =    0x8, /*   4 */
 320   STAT_TYPE_BLOCKS_NO_PULL            =   0x10, /*   5 */
 321   STAT_TYPE_BLOCKS_MANY_PUSH_NO_PULL  =   0x20, /*   6 */
 322   STAT_TYPE_BLOCKS_NO_PUSH_NO_PULL    =   0x40, /*   7 */
 323   STAT_TYPE_ISSUED_PUSH_SEND          =   0x80, /*   8 */
 324   STAT_TYPE_ISSUED_PULL_REQ           =  0x100, /*   9 */
 325   STAT_TYPE_ISSUED_PULL_REP           =  0x200, /*  10 */
 326   STAT_TYPE_SENT_PUSH_SEND            =  0x400, /*  11 */
 327   STAT_TYPE_SENT_PULL_REQ             =  0x800, /*  12 */
 328   STAT_TYPE_SENT_PULL_REP             = 0x1000, /*  13 */
 329   STAT_TYPE_RECV_PUSH_SEND            = 0x2000, /*  14 */
 330   STAT_TYPE_RECV_PULL_REQ             = 0x4000, /*  15 */
 331   STAT_TYPE_RECV_PULL_REP             = 0x8000, /*  16 */
 332   STAT_TYPE_MAX          = 0x80000000, /*  32 */
 333 };
 334
 335 struct STATcls
 336 {
 337   struct RPSPeer *rps_peer;
 338   enum STAT_TYPE stat_type;
 339 };
 340
 341
 342 /**
 343  * Information for all the peers.
 344  */
 345 static struct RPSPeer *rps_peers;
 346
 347 /**
 348  * Peermap to get the index of a given peer ID quick.
 349  */
 350 static struct GNUNET_CONTAINER_MultiPeerMap *peer_map;
 351
 352 /**
 353  * IDs of the peers.
 354  */
 355 static struct GNUNET_PeerIdentity *rps_peer_ids;
 356
 357 /**
 358  * ID of the targeted peer.
 359  */
 360 static struct GNUNET_PeerIdentity *target_peer;
 361
 362 /**
 363  * ID of the peer that requests for the evaluation.
 364  */
 365 static struct RPSPeer *eval_peer;
 366
 367 /**
 368  * Number of online peers.
 369  */
 370 static unsigned int num_peers_online;
 371
 372 /**
 373  * @brief The added sizes of the peer's views
 374  */
 375 static unsigned int view_sizes;
 376
 377 /**
 378  * Return value from 'main'.
 379  */
 380 static int ok;
 381
 382 /**
 383  * Identifier for the churn task that runs periodically
 384  */
 385 static struct GNUNET_SCHEDULER_Task *post_test_task;
 386
 387 /**
 388  * Identifier for the churn task that runs periodically
 389  */
 390 static struct GNUNET_SCHEDULER_Task *shutdown_task;
 391
 392 /**
 393  * Identifier for the churn task that runs periodically
 394  */
 395 static struct GNUNET_SCHEDULER_Task *churn_task;
 396
 397 /**
 398  * Called to initialise the given RPSPeer
 399  */
 400 typedef void (*InitPeer) (struct RPSPeer *rps_peer);
 401
 402 /**
 403  * @brief Called directly after connecting to the service
 404  *
 405  * @param rps_peer Specific peer the function is called on
 406  * @param h the handle to the rps service
 407  */
 408 typedef void (*PreTest) (struct RPSPeer *rps_peer, struct GNUNET_RPS_Handle *h);
 409
 410 /**
 411  * @brief Executes functions to test the api/service for a given peer
 412  *
 413  * Called from within #rps_connect_complete_cb ()
 414  * Implemented by #churn_test_cb, #profiler_cb, #mal_cb, #single_req_cb,
 415  * #delay_req_cb, #seed_big_cb, #single_peer_seed_cb, #seed_cb, #req_cancel_cb
 416  *
 417  * @param rps_peer the peer the task runs on
 418  */
 419 typedef void (*MainTest) (struct RPSPeer *rps_peer);
 420
 421 /**
 422  * Callback called once the requested random peers are available
 423  */
 424 typedef void (*ReplyHandle) (void *cls,
 425                              uint64_t n,
 426                              const struct GNUNET_PeerIdentity *recv_peers);
 427
 428 /**
 429  * Called directly before disconnecting from the service
 430  */
 431 typedef void (*PostTest) (struct RPSPeer *peer);
 432
 433 /**
 434  * Function called after disconnect to evaluate test success
 435  */
 436 typedef int (*EvaluationCallback) (void);
 437
 438 /**
 439  * @brief Do we have Churn?
 440  */
 441 enum OPTION_CHURN {
 442   /**
 443    * @brief If we have churn this is set
 444    */
 445   HAVE_CHURN,
 446   /**
 447    * @brief If we have no churn this is set
 448    */
 449   HAVE_NO_CHURN,
 450 };
 451
 452 /**
 453  * @brief Is it ok to quit the test before the timeout?
 454  */
 455 enum OPTION_QUICK_QUIT {
 456   /**
 457    * @brief It is ok for the test to quit before the timeout triggers
 458    */
 459   HAVE_QUICK_QUIT,
 460
 461   /**
 462    * @brief It is NOT ok for the test to quit before the timeout triggers
 463    */
 464   HAVE_NO_QUICK_QUIT,
 465 };
 466
 467 /**
 468  * @brief Do we collect statistics at the end?
 469  */
 470 enum OPTION_COLLECT_STATISTICS {
 471   /**
 472    * @brief We collect statistics at the end
 473    */
 474   COLLECT_STATISTICS,
 475
 476   /**
 477    * @brief We do not collect statistics at the end
 478    */
 479   NO_COLLECT_STATISTICS,
 480 };
 481
 482 /**
 483  * @brief Do we collect views during run?
 484  */
 485 enum OPTION_COLLECT_VIEW {
 486   /**
 487    * @brief We collect view during run
 488    */
 489   COLLECT_VIEW,
 490
 491   /**
 492    * @brief We do not collect the view during run
 493    */
 494   NO_COLLECT_VIEW,
 495 };
 496
 497 /**
 498  * Structure to define a single test
 499  */
 500 struct SingleTestRun
 501 {
 502   /**
 503    * Name of the test
 504    */
 505   char *name;
 506
 507   /**
 508    * Called with a single peer in order to initialise that peer
 509    */
 510   InitPeer init_peer;
 511
 512   /**
 513    * Called directly after connecting to the service
 514    */
 515   PreTest pre_test;
 516
 517   /**
 518    * Main function for each peer
 519    */
 520   MainTest main_test;
 521
 522   /**
 523    * Callback called once the requested peers are available
 524    */
 525   ReplyHandle reply_handle;
 526
 527   /**
 528    * Called directly before disconnecting from the service
 529    */
 530   PostTest post_test;
 531
 532   /**
 533    * Function to evaluate the test results
 534    */
 535   EvaluationCallback eval_cb;
 536
 537   /**
 538    * Request interval
 539    */
 540   uint32_t request_interval;
 541
 542   /**
 543    * Number of Requests to make.
 544    */
 545   uint32_t num_requests;
 546
 547   /**
 548    * Run with (-out) churn
 549    */
 550   enum OPTION_CHURN have_churn;
 551
 552   /**
 553    * Quit test before timeout?
 554    */
 555   enum OPTION_QUICK_QUIT have_quick_quit;
 556
 557   /**
 558    * Collect statistics at the end?
 559    */
 560   enum OPTION_COLLECT_STATISTICS have_collect_statistics;
 561
 562   /**
 563    * Collect view during run?
 564    */
 565   enum OPTION_COLLECT_VIEW have_collect_view;
 566
 567   /**
 568    * @brief Mark which values from the statistics service to collect at the end
 569    * of the run
 570    */
 571   uint32_t stat_collect_flags;
 572 } cur_test_run;
 573
 574 /**
 575  * Did we finish the test?
 576  */
 577 static int post_test;
 578
 579 /**
 580  * Are we shutting down?
 581  */
 582 static int in_shutdown;
 583
 584 /**
 585  * Append arguments to file
 586  */
 587 static void
 588 tofile_ (const char *file_name, const char *line)
 589 {
 590   struct GNUNET_DISK_FileHandle *f;
 591   /* char output_buffer[512]; */
 592   size_t size;
 593   /* int size; */
 594   size_t size2;
 595
 596   if (NULL == (f = GNUNET_DISK_file_open (file_name,
 597                                           GNUNET_DISK_OPEN_APPEND |
 598                                           GNUNET_DISK_OPEN_WRITE |
 599                                           GNUNET_DISK_OPEN_CREATE,
 600                                           GNUNET_DISK_PERM_USER_READ |
 601                                           GNUNET_DISK_PERM_USER_WRITE |
 602                                           GNUNET_DISK_PERM_GROUP_READ |
 603                                           GNUNET_DISK_PERM_OTHER_READ)))
 604   {
 605     GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
 606                 "Not able to open file %s\n",
 607                 file_name);
 608     return;
 609   }
 610   /* size = GNUNET_snprintf (output_buffer,
 611                           sizeof (output_buffer),
 612                           "%llu %s\n",
 613                           GNUNET_TIME_absolute_get ().abs_value_us,
 614                           line);
 615   if (0 > size)
 616   {
 617     GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
 618                 "Failed to write string to buffer (size: %i)\n",
 619                 size);
 620     return;
 621   } */
 622
 623   size = strlen (line) * sizeof (char);
 624
 625   size2 = GNUNET_DISK_file_write (f, line, size);
 626   if (size != size2)
 627   {
 628     GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
 629                 "Unable to write to file! (Size: %lu, size2: %lu)\n",
 630                 size,
 631                 size2);
 632     if (GNUNET_YES != GNUNET_DISK_file_close (f))
 633     {
 634       GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
 635                   "Unable to close file\n");
 636     }
 637     return;
 638   }
 639
 640   if (GNUNET_YES != GNUNET_DISK_file_close (f))
 641   {
 642     GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
 643                 "Unable to close file\n");
 644   }
 645 }
 646
 647 /**
 648  * This function is used to facilitate writing important information to disk
 649  */
 650 #define tofile(file_name, ...) do {\
 651   char tmp_buf[512];\
 652     int size;\
 653     size = GNUNET_snprintf(tmp_buf,sizeof(tmp_buf),__VA_ARGS__);\
 654     if (0 > size)\
 655       GNUNET_log (GNUNET_ERROR_TYPE_WARNING,\
 656                      "Failed to create tmp_buf\n");\
 657     else\
 658       tofile_(file_name,tmp_buf);\
 659   } while (0);
 660
 661
 662 /**
 663  * Write the ids and their according index in the given array to a file
 664  * Unused
 665  */
 666 /* static void
 667 ids_to_file (char *file_name,
 668              struct GNUNET_PeerIdentity *peer_ids,
 669              unsigned int num_peer_ids)
 670 {
 671   unsigned int i;
 672
 673   for (i=0 ; i < num_peer_ids ; i++)
 674   {
 675     to_file (file_name,
 676              "%u\t%s",
 677              i,
 678              GNUNET_i2s_full (&peer_ids[i]));
 679   }
 680 } */
 681
 682 /**
 683  * Task run on timeout to collect statistics and potentially shut down.
 684  */
 685 static void
 686 post_test_op (void *cls);
 687
 688
 689 /**
 690  * Test the success of a single test
 691  */
 692 static int
 693 evaluate (void)
 694 {
 695   unsigned int i;
 696   int tmp_ok;
 697
 698   tmp_ok = 1;
 699
 700   for (i = 0; i < num_peers; i++)
 701   {
 702     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 703         "%u. peer [%s] received %u of %u expected peer_ids: %i\n",
 704         i,
 705         GNUNET_i2s (rps_peers[i].peer_id),
 706         rps_peers[i].num_recv_ids,
 707         rps_peers[i].num_ids_to_request,
 708         (rps_peers[i].num_ids_to_request == rps_peers[i].num_recv_ids));
 709     tmp_ok &= (rps_peers[i].num_ids_to_request == rps_peers[i].num_recv_ids);
 710   }
 711   return tmp_ok? 0 : 1;
 712 }
 713
 714
 715 /**
 716  * Creates an oplist entry and adds it to the oplist DLL
 717  */
 718 static struct OpListEntry *
 719 make_oplist_entry ()
 720 {
 721   struct OpListEntry *entry;
 722
 723   entry = GNUNET_new (struct OpListEntry);
 724   GNUNET_CONTAINER_DLL_insert_tail (oplist_head, oplist_tail, entry);
 725   return entry;
 726 }
 727
 728
 729 /**
 730  * @brief Checks if given peer already received its statistics value from the
 731  * statistics service.
 732  *
 733  * @param rps_peer the peer to check for
 734  *
 735  * @return #GNUNET_YES if so
 736  *         #GNUNET_NO otherwise
 737  */
 738 static int check_statistics_collect_completed_single_peer (
 739     const struct RPSPeer *rps_peer)
 740 {
 741   if (cur_test_run.stat_collect_flags !=
 742         (cur_test_run.stat_collect_flags &
 743           rps_peer->stat_collected_flags))
 744   {
 745     return GNUNET_NO;
 746   }
 747   return GNUNET_YES;
 748 }
 749
 750
 751 /**
 752  * @brief Checks if all peers already received their statistics value from the
 753  * statistics service.
 754  *
 755  * @return #GNUNET_YES if so
 756  *         #GNUNET_NO otherwise
 757  */
 758 static int check_statistics_collect_completed ()
 759 {
 760   uint32_t i;
 761
 762   for (i = 0; i < num_peers; i++)
 763   {
 764     if (GNUNET_NO == check_statistics_collect_completed_single_peer (&rps_peers[i]))
 765     {
 766       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 767           "At least Peer %" PRIu32 " did not yet receive all statistics values\n",
 768           i);
 769       return GNUNET_NO;
 770     }
 771   }
 772   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 773       "All peers received their statistics values\n");
 774   return GNUNET_YES;
 775 }
 776
 777
 778 /**
 779  * Task run on timeout to shut everything down.
 780  */
 781 static void
 782 shutdown_op (void *cls)
 783 {
 784   unsigned int i;
 785   (void) cls;
 786
 787   GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
 788               "Shutdown task scheduled, going down.\n");
 789   in_shutdown = GNUNET_YES;
 790   if (NULL != post_test_task)
 791   {
 792     GNUNET_SCHEDULER_cancel (post_test_task);
 793     post_test_op (NULL);
 794   }
 795   if (NULL != churn_task)
 796   {
 797     GNUNET_SCHEDULER_cancel (churn_task);
 798     churn_task = NULL;
 799   }
 800   for (i = 0; i < num_peers; i++)
 801   {
 802     if (NULL != rps_peers[i].rps_handle)
 803     {
 804       GNUNET_RPS_disconnect (rps_peers[i].rps_handle);
 805     }
 806     if (NULL != rps_peers[i].op)
 807     {
 808       GNUNET_TESTBED_operation_done (rps_peers[i].op);
 809     }
 810   }
 811 }
 812
 813
 814 /**
 815  * Task run on timeout to collect statistics and potentially shut down.
 816  */
 817 static void
 818 post_test_op (void *cls)
 819 {
 820   unsigned int i;
 821   (void) cls;
 822
 823   post_test_task = NULL;
 824   post_test = GNUNET_YES;
 825   GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
 826               "Post test task scheduled, going down.\n");
 827   if (NULL != churn_task)
 828   {
 829     GNUNET_SCHEDULER_cancel (churn_task);
 830     churn_task = NULL;
 831   }
 832   for (i = 0; i < num_peers; i++)
 833   {
 834     if (NULL != cur_test_run.post_test)
 835     {
 836       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Executing post_test for peer %u\n", i);
 837       cur_test_run.post_test (&rps_peers[i]);
 838     }
 839     if (NULL != rps_peers[i].op)
 840     {
 841       GNUNET_TESTBED_operation_done (rps_peers[i].op);
 842       rps_peers[i].op = NULL;
 843     }
 844   }
 845   /* If we do not collect statistics, shut down directly */
 846   if (NO_COLLECT_STATISTICS == cur_test_run.have_collect_statistics ||
 847       GNUNET_YES == check_statistics_collect_completed())
 848   {
 849     GNUNET_SCHEDULER_shutdown ();
 850   }
 851 }
 852
 853
 854 /**
 855  * Seed peers.
 856  */
 857 static void
 858 seed_peers (void *cls)
 859 {
 860   struct RPSPeer *peer = cls;
 861   unsigned int amount;
 862   unsigned int i;
 863
 864   if (GNUNET_YES == in_shutdown || GNUNET_YES == post_test)
 865   {
 866     return;
 867   }
 868
 869   GNUNET_assert (NULL != peer->rps_handle);
 870
 871   // TODO if malicious don't seed mal peers
 872   amount = round (.5 * num_peers);
 873
 874   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Seeding peers:\n");
 875   for (i = 0 ; i < amount ; i++)
 876     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Seeding %u. peer: %s\n",
 877                 i,
 878                 GNUNET_i2s (&rps_peer_ids[i]));
 879
 880   GNUNET_RPS_seed_ids (peer->rps_handle, amount, rps_peer_ids);
 881 }
 882
 883
 884 /**
 885  * Seed peers.
 886  */
 887 static void
 888 seed_peers_big (void *cls)
 889 {
 890   struct RPSPeer *peer = cls;
 891   unsigned int seed_msg_size;
 892   uint32_t num_peers_max;
 893   unsigned int amount;
 894   unsigned int i;
 895
 896   seed_msg_size = 8; /* sizeof (struct GNUNET_RPS_CS_SeedMessage) */
 897   num_peers_max = (GNUNET_MAX_MESSAGE_SIZE - seed_msg_size) /
 898     sizeof (struct GNUNET_PeerIdentity);
 899   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 900       "Peers that fit in one seed msg; %u\n",
 901       num_peers_max);
 902   amount = num_peers_max + (0.5 * num_peers_max);
 903   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 904       "Seeding many (%u) peers:\n",
 905       amount);
 906   struct GNUNET_PeerIdentity ids_to_seed[amount];
 907   for (i = 0; i < amount; i++)
 908   {
 909     ids_to_seed[i] = *peer->peer_id;
 910     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Seeding %u. peer: %s\n",
 911                 i,
 912                 GNUNET_i2s (&ids_to_seed[i]));
 913   }
 914
 915   GNUNET_RPS_seed_ids (peer->rps_handle, amount, ids_to_seed);
 916 }
 917
 918 /**
 919  * Get the id of peer i.
 920  */
 921   void
 922 info_cb (void *cb_cls,
 923          struct GNUNET_TESTBED_Operation *op,
 924          const struct GNUNET_TESTBED_PeerInformation *pinfo,
 925          const char *emsg)
 926 {
 927   struct OpListEntry *entry = (struct OpListEntry *) cb_cls;
 928   (void) op;
 929
 930   if (GNUNET_YES == in_shutdown || GNUNET_YES == post_test)
 931   {
 932     return;
 933   }
 934
 935   if (NULL == pinfo || NULL != emsg)
 936   {
 937     GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "Got Error: %s\n", emsg);
 938     GNUNET_TESTBED_operation_done (entry->op);
 939     return;
 940   }
 941
 942   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 943               "Peer %u is %s\n",
 944               entry->index,
 945               GNUNET_i2s (pinfo->result.id));
 946
 947   rps_peer_ids[entry->index] = *(pinfo->result.id);
 948   rps_peers[entry->index].peer_id = &rps_peer_ids[entry->index];
 949
 950   GNUNET_assert (GNUNET_OK ==
 951       GNUNET_CONTAINER_multipeermap_put (peer_map,
 952         &rps_peer_ids[entry->index],
 953         &rps_peers[entry->index],
 954         GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
 955   tofile ("/tmp/rps/peer_ids",
 956            "%u\t%s\n",
 957            entry->index,
 958            GNUNET_i2s_full (&rps_peer_ids[entry->index]));
 959
 960   GNUNET_CONTAINER_DLL_remove (oplist_head, oplist_tail, entry);
 961   GNUNET_TESTBED_operation_done (entry->op);
 962   GNUNET_free (entry);
 963 }
 964
 965
 966 /**
 967  * Callback to be called when RPS service connect operation is completed
 968  *
 969  * @param cls the callback closure from functions generating an operation
 970  * @param op the operation that has been finished
 971  * @param ca_result the RPS service handle returned from rps_connect_adapter
 972  * @param emsg error message in case the operation has failed; will be NULL if
 973  *          operation has executed successfully.
 974  */
 975 static void
 976 rps_connect_complete_cb (void *cls,
 977                          struct GNUNET_TESTBED_Operation *op,
 978                          void *ca_result,
 979                          const char *emsg)
 980 {
 981   struct RPSPeer *rps_peer = cls;
 982   struct GNUNET_RPS_Handle *rps = ca_result;
 983
 984   GNUNET_assert (NULL != ca_result);
 985
 986   if (GNUNET_YES == in_shutdown || GNUNET_YES == post_test)
 987   {
 988     return;
 989   }
 990
 991   rps_peer->rps_handle = rps;
 992   rps_peer->online = GNUNET_YES;
 993   num_peers_online++;
 994
 995   GNUNET_assert (op == rps_peer->op);
 996   if (NULL != emsg)
 997   {
 998     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 999                 "Failed to connect to RPS service: %s\n",
1000                 emsg);
1001     ok = 1;
1002     GNUNET_SCHEDULER_shutdown ();
1003     return;
1004   }
1005
1006   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Started client successfully\n");
1007
1008   cur_test_run.main_test (rps_peer);
1009 }
1010
1011
1012 /**
1013  * Adapter function called to establish a connection to
1014  * the RPS service.
1015  *
1016  * @param cls closure
1017  * @param cfg configuration of the peer to connect to; will be available until
1018  *          GNUNET_TESTBED_operation_done() is called on the operation returned
1019  *          from GNUNET_TESTBED_service_connect()
1020  * @return service handle to return in 'op_result', NULL on error
1021  */
1022 static void *
1023 rps_connect_adapter (void *cls,
1024                                  const struct GNUNET_CONFIGURATION_Handle *cfg)
1025 {
1026   struct GNUNET_RPS_Handle *h;
1027
1028   h = GNUNET_RPS_connect (cfg);
1029   GNUNET_assert (NULL != h);
1030
1031   if (NULL != cur_test_run.pre_test)
1032     cur_test_run.pre_test (cls, h);
1033   GNUNET_assert (NULL != h);
1034
1035   return h;
1036 }
1037
1038 /**
1039  * Called to open a connection to the peer's statistics
1040  *
1041  * @param cls peer context
1042  * @param cfg configuration of the peer to connect to; will be available until
1043  *          GNUNET_TESTBED_operation_done() is called on the operation returned
1044  *          from GNUNET_TESTBED_service_connect()
1045  * @return service handle to return in 'op_result', NULL on error
1046  */
1047 static void *
1048 stat_connect_adapter (void *cls,
1049                       const struct GNUNET_CONFIGURATION_Handle *cfg)
1050 {
1051   struct RPSPeer *peer = cls;
1052
1053   peer->stats_h = GNUNET_STATISTICS_create ("rps-profiler", cfg);
1054   return peer->stats_h;
1055 }
1056
1057 /**
1058  * Called to disconnect from peer's statistics service
1059  *
1060  * @param cls peer context
1061  * @param op_result service handle returned from the connect adapter
1062  */
1063 static void
1064 stat_disconnect_adapter (void *cls, void *op_result)
1065 {
1066   struct RPSPeer *peer = cls;
1067
1068   //GNUNET_break (GNUNET_OK == GNUNET_STATISTICS_watch_cancel
1069   //              (peer->stats_h, "core", "# peers connected",
1070   //               stat_iterator, peer));
1071   //GNUNET_break (GNUNET_OK == GNUNET_STATISTICS_watch_cancel
1072   //              (peer->stats_h, "nse", "# peers connected",
1073   //               stat_iterator, peer));
1074   GNUNET_STATISTICS_destroy (op_result, GNUNET_NO);
1075   peer->stats_h = NULL;
1076 }
1077
1078 /**
1079  * Called after successfully opening a connection to a peer's statistics
1080  * service; we register statistics monitoring for CORE and NSE here.
1081  *
1082  * @param cls the callback closure from functions generating an operation
1083  * @param op the operation that has been finished
1084  * @param ca_result the service handle returned from GNUNET_TESTBED_ConnectAdapter()
1085  * @param emsg error message in case the operation has failed; will be NULL if
1086  *          operation has executed successfully.
1087  */
1088 static void
1089 stat_complete_cb (void *cls, struct GNUNET_TESTBED_Operation *op,
1090                   void *ca_result, const char *emsg )
1091 {
1092   //struct GNUNET_STATISTICS_Handle *sh = ca_result;
1093   //struct RPSPeer *peer = (struct RPSPeer *) cls;
1094   (void) cls;
1095   (void) op;
1096   (void) ca_result;
1097
1098   if (NULL != emsg)
1099   {
1100     GNUNET_break (0);
1101     return;
1102   }
1103   //GNUNET_break (GNUNET_OK == GNUNET_STATISTICS_watch
1104   //              (sh, "core", "# peers connected",
1105   //               stat_iterator, peer));
1106   //GNUNET_break (GNUNET_OK == GNUNET_STATISTICS_watch
1107   //              (sh, "nse", "# peers connected",
1108   //               stat_iterator, peer));
1109 }
1110
1111
1112 /**
1113  * Adapter function called to destroy connection to
1114  * RPS service.
1115  *
1116  * @param cls closure
1117  * @param op_result service handle returned from the connect adapter
1118  */
1119 static void
1120 rps_disconnect_adapter (void *cls,
1121                                           void *op_result)
1122 {
1123   struct RPSPeer *peer = cls;
1124   struct GNUNET_RPS_Handle *h = op_result;
1125
1126   if (NULL != peer->rps_srh)
1127   {
1128     GNUNET_RPS_stream_cancel (peer->rps_srh);
1129     peer->rps_srh = NULL;
1130   }
1131   GNUNET_assert (NULL != peer);
1132   GNUNET_RPS_disconnect (h);
1133   peer->rps_handle = NULL;
1134 }
1135
1136
1137 /***********************************************************************
1138  * Definition of tests
1139 ***********************************************************************/
1140
1141 // TODO check whether tests can be stopped earlier
1142 static int
1143 default_eval_cb (void)
1144 {
1145   return evaluate ();
1146 }
1147
1148 static int
1149 no_eval (void)
1150 {
1151   return 0;
1152 }
1153
1154 /**
1155  * Initialise given RPSPeer
1156  */
1157 static void default_init_peer (struct RPSPeer *rps_peer)
1158 {
1159   rps_peer->num_ids_to_request = 1;
1160 }
1161
1162 /**
1163  * Callback to call on receipt of a reply
1164  *
1165  * @param cls closure
1166  * @param n number of peers
1167  * @param recv_peers the received peers
1168  */
1169 static void
1170 default_reply_handle (void *cls,
1171                       uint64_t n,
1172                       const struct GNUNET_PeerIdentity *recv_peers)
1173 {
1174   struct RPSPeer *rps_peer;
1175   struct PendingReply *pending_rep = (struct PendingReply *) cls;
1176   unsigned int i;
1177
1178   rps_peer = pending_rep->rps_peer;
1179   GNUNET_CONTAINER_DLL_remove (rps_peer->pending_rep_head,
1180                                rps_peer->pending_rep_tail,
1181                                pending_rep);
1182   rps_peer->num_pending_reps--;
1183   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1184               "[%s] got %" PRIu64 " peers:\n",
1185               GNUNET_i2s (rps_peer->peer_id),
1186               n);
1187
1188   for (i = 0; i < n; i++)
1189   {
1190     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1191                 "%u: %s\n",
1192                 i,
1193                 GNUNET_i2s (&recv_peers[i]));
1194
1195     rps_peer->num_recv_ids++;
1196   }
1197
1198   if (0 == evaluate () && HAVE_QUICK_QUIT == cur_test_run.have_quick_quit)
1199   {
1200     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Test succeeded before timeout\n");
1201     GNUNET_assert (NULL != post_test_task);
1202     GNUNET_SCHEDULER_cancel (post_test_task);
1203     post_test_task = GNUNET_SCHEDULER_add_now (&post_test_op, NULL);
1204     GNUNET_assert (NULL!= post_test_task);
1205   }
1206 }
1207
1208 /**
1209  * Request random peers.
1210  */
1211 static void
1212 request_peers (void *cls)
1213 {
1214   struct PendingRequest *pending_req = cls;
1215   struct RPSPeer *rps_peer;
1216   struct PendingReply *pending_rep;
1217
1218   if (GNUNET_YES == in_shutdown || GNUNET_YES == post_test)
1219     return;
1220   rps_peer = pending_req->rps_peer;
1221   GNUNET_assert (1 <= rps_peer->num_pending_reqs);
1222   GNUNET_CONTAINER_DLL_remove (rps_peer->pending_req_head,
1223                                rps_peer->pending_req_tail,
1224                                pending_req);
1225   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1226               "Requesting one peer\n");
1227   pending_rep = GNUNET_new (struct PendingReply);
1228   pending_rep->rps_peer = rps_peer;
1229   pending_rep->req_handle = GNUNET_RPS_request_peers (rps_peer->rps_handle,
1230       1,
1231       cur_test_run.reply_handle,
1232       pending_rep);
1233   GNUNET_CONTAINER_DLL_insert_tail (rps_peer->pending_rep_head,
1234                                     rps_peer->pending_rep_tail,
1235                                     pending_rep);
1236   rps_peer->num_pending_reps++;
1237   rps_peer->num_pending_reqs--;
1238 }
1239
1240 static void
1241 cancel_pending_req (struct PendingRequest *pending_req)
1242 {
1243   struct RPSPeer *rps_peer;
1244
1245   rps_peer = pending_req->rps_peer;
1246   GNUNET_CONTAINER_DLL_remove (rps_peer->pending_req_head,
1247                                rps_peer->pending_req_tail,
1248                                pending_req);
1249   rps_peer->num_pending_reqs--;
1250   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1251               "Cancelling pending request\n");
1252   GNUNET_SCHEDULER_cancel (pending_req->request_task);
1253   GNUNET_free (pending_req);
1254 }
1255
1256 static void
1257 cancel_request (struct PendingReply *pending_rep)
1258 {
1259   struct RPSPeer *rps_peer;
1260
1261   rps_peer = pending_rep->rps_peer;
1262   GNUNET_CONTAINER_DLL_remove (rps_peer->pending_rep_head,
1263                                rps_peer->pending_rep_tail,
1264                                pending_rep);
1265   rps_peer->num_pending_reps--;
1266   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1267               "Cancelling request\n");
1268   GNUNET_RPS_request_cancel (pending_rep->req_handle);
1269   GNUNET_free (pending_rep);
1270 }
1271
1272 /**
1273  * Cancel a request.
1274  */
1275 static void
1276 cancel_request_cb (void *cls)
1277 {
1278   struct RPSPeer *rps_peer = cls;
1279   struct PendingReply *pending_rep;
1280
1281   if (GNUNET_YES == in_shutdown || GNUNET_YES == post_test)
1282     return;
1283   pending_rep = rps_peer->pending_rep_head;
1284   GNUNET_assert (1 <= rps_peer->num_pending_reps);
1285   cancel_request (pending_rep);
1286 }
1287
1288
1289 /**
1290  * Schedule requests for peer @a rps_peer that have neither been scheduled, nor
1291  * issued, nor replied
1292  */
1293 void
1294 schedule_missing_requests (struct RPSPeer *rps_peer)
1295 {
1296   unsigned int i;
1297   struct PendingRequest *pending_req;
1298
1299   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1300       "Scheduling %u - %u missing requests\n",
1301       rps_peer->num_ids_to_request,
1302       rps_peer->num_pending_reqs + rps_peer->num_pending_reps);
1303   GNUNET_assert (rps_peer->num_pending_reqs + rps_peer->num_pending_reps <=
1304       rps_peer->num_ids_to_request);
1305   for (i = rps_peer->num_pending_reqs + rps_peer->num_pending_reps;
1306        i < rps_peer->num_ids_to_request; i++)
1307   {
1308     pending_req = GNUNET_new (struct PendingRequest);
1309     pending_req->rps_peer = rps_peer;
1310     pending_req->request_task = GNUNET_SCHEDULER_add_delayed (
1311         GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS,
1312           cur_test_run.request_interval * i),
1313         request_peers,
1314         pending_req);
1315     GNUNET_CONTAINER_DLL_insert_tail (rps_peer->pending_req_head,
1316                                       rps_peer->pending_req_tail,
1317                                       pending_req);
1318     rps_peer->num_pending_reqs++;
1319   }
1320 }
1321
1322 void
1323 cancel_pending_req_rep (struct RPSPeer *rps_peer)
1324 {
1325   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1326       "Cancelling all (pending) requests.\n");
1327   while (NULL != rps_peer->pending_req_head)
1328     cancel_pending_req (rps_peer->pending_req_head);
1329   GNUNET_assert (0 == rps_peer->num_pending_reqs);
1330   while (NULL != rps_peer->pending_rep_head)
1331     cancel_request (rps_peer->pending_rep_head);
1332   GNUNET_assert (0 == rps_peer->num_pending_reps);
1333 }
1334
1335 /***********************************
1336  * MALICIOUS
1337 ***********************************/
1338
1339 /**
1340  * Initialise only non-mal RPSPeers
1341  */
1342 static void mal_init_peer (struct RPSPeer *rps_peer)
1343 {
1344   if (rps_peer->index >= round (portion * num_peers))
1345     rps_peer->num_ids_to_request = 1;
1346 }
1347
1348
1349 /**
1350  * @brief Set peers to (non-)malicious before execution
1351  *
1352  * Of signature #PreTest
1353  *
1354  * @param rps_peer the peer to set (non-) malicious
1355  * @param h the handle to the service
1356  */
1357 static void
1358 mal_pre (struct RPSPeer *rps_peer, struct GNUNET_RPS_Handle *h)
1359 {
1360   #if ENABLE_MALICIOUS
1361   uint32_t num_mal_peers;
1362
1363   GNUNET_assert ( (1 >= portion) &&
1364                   (0 <  portion) );
1365   num_mal_peers = round (portion * num_peers);
1366
1367   if (rps_peer->index < num_mal_peers)
1368   {
1369     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1370                 "%u. peer [%s] of %" PRIu32 " malicious peers turning malicious\n",
1371                 rps_peer->index,
1372                 GNUNET_i2s (rps_peer->peer_id),
1373                 num_mal_peers);
1374
1375     GNUNET_RPS_act_malicious (h, mal_type, num_mal_peers,
1376                               rps_peer_ids, target_peer);
1377   }
1378   #endif /* ENABLE_MALICIOUS */
1379 }
1380
1381 static void
1382 mal_cb (struct RPSPeer *rps_peer)
1383 {
1384   uint32_t num_mal_peers;
1385
1386   if (GNUNET_YES == in_shutdown || GNUNET_YES == post_test)
1387   {
1388     return;
1389   }
1390
1391   #if ENABLE_MALICIOUS
1392   GNUNET_assert ( (1 >= portion) &&
1393                   (0 <  portion) );
1394   num_mal_peers = round (portion * num_peers);
1395
1396   if (rps_peer->index >= num_mal_peers)
1397   { /* It's useless to ask a malicious peer about a random sample -
1398        it's not sampling */
1399     GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 2),
1400                                   seed_peers, rps_peer);
1401     schedule_missing_requests (rps_peer);
1402   }
1403   #endif /* ENABLE_MALICIOUS */
1404 }
1405
1406
1407 /***********************************
1408  * SINGLE_REQUEST
1409 ***********************************/
1410 static void
1411 single_req_cb (struct RPSPeer *rps_peer)
1412 {
1413   if (GNUNET_YES == in_shutdown || GNUNET_YES == post_test)
1414   {
1415     return;
1416   }
1417
1418   schedule_missing_requests (rps_peer);
1419 }
1420
1421 /***********************************
1422  * DELAYED_REQUESTS
1423 ***********************************/
1424 static void
1425 delay_req_cb (struct RPSPeer *rps_peer)
1426 {
1427   if (GNUNET_YES == in_shutdown || GNUNET_YES == post_test)
1428   {
1429     return;
1430   }
1431
1432   schedule_missing_requests (rps_peer);
1433 }
1434
1435 /***********************************
1436  * SEED
1437 ***********************************/
1438 static void
1439 seed_cb (struct RPSPeer *rps_peer)
1440 {
1441   if (GNUNET_YES == in_shutdown || GNUNET_YES == post_test)
1442   {
1443     return;
1444   }
1445
1446   GNUNET_SCHEDULER_add_delayed (
1447       GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 10),
1448       seed_peers, rps_peer);
1449 }
1450
1451 /***********************************
1452  * SEED_BIG
1453 ***********************************/
1454 static void
1455 seed_big_cb (struct RPSPeer *rps_peer)
1456 {
1457   if (GNUNET_YES == in_shutdown || GNUNET_YES == post_test)
1458   {
1459     return;
1460   }
1461
1462   // TODO test seeding > GNUNET_MAX_MESSAGE_SIZE peers
1463   GNUNET_SCHEDULER_add_delayed (
1464       GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 2),
1465       seed_peers_big, rps_peer);
1466 }
1467
1468 /***********************************
1469  * SINGLE_PEER_SEED
1470 ***********************************/
1471 static void
1472 single_peer_seed_cb (struct RPSPeer *rps_peer)
1473 {
1474   (void) rps_peer;
1475   // TODO
1476 }
1477
1478 /***********************************
1479  * SEED_REQUEST
1480 ***********************************/
1481 static void
1482 seed_req_cb (struct RPSPeer *rps_peer)
1483 {
1484   if (GNUNET_YES == in_shutdown || GNUNET_YES == post_test)
1485   {
1486     return;
1487   }
1488
1489   GNUNET_SCHEDULER_add_delayed (
1490       GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 2),
1491       seed_peers, rps_peer);
1492   schedule_missing_requests (rps_peer);
1493 }
1494
1495 //TODO start big mal
1496
1497 /***********************************
1498  * REQUEST_CANCEL
1499 ***********************************/
1500 static void
1501 req_cancel_cb (struct RPSPeer *rps_peer)
1502 {
1503   if (GNUNET_YES == in_shutdown || GNUNET_YES == post_test)
1504   {
1505     return;
1506   }
1507
1508   schedule_missing_requests (rps_peer);
1509   GNUNET_SCHEDULER_add_delayed (
1510       GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS,
1511                                      (cur_test_run.request_interval + 1)),
1512       cancel_request_cb, rps_peer);
1513 }
1514
1515 /***********************************
1516  * CHURN
1517 ***********************************/
1518
1519 static void
1520 churn (void *cls);
1521
1522 /**
1523  * @brief Starts churn
1524  *
1525  * Has signature of #MainTest
1526  *
1527  * This is not implemented too nicely as this is called for each peer, but we
1528  * only need to call it once. (Yes we check that we only schedule the task
1529  * once.)
1530  *
1531  * @param rps_peer The peer it's called for
1532  */
1533 static void
1534 churn_test_cb (struct RPSPeer *rps_peer)
1535 {
1536   if (GNUNET_YES == in_shutdown || GNUNET_YES == post_test)
1537   {
1538     return;
1539   }
1540
1541   /* Start churn */
1542   if (HAVE_CHURN == cur_test_run.have_churn && NULL == churn_task)
1543   {
1544     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1545                 "Starting churn task\n");
1546     churn_task = GNUNET_SCHEDULER_add_delayed (
1547           GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 5),
1548           churn,
1549           NULL);
1550   } else {
1551     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1552                 "Not starting churn task\n");
1553   }
1554
1555   schedule_missing_requests (rps_peer);
1556 }
1557
1558 /***********************************
1559  * SUB
1560 ***********************************/
1561
1562 static void
1563 got_stream_peer_cb (void *cls,
1564                     uint64_t num_peers,
1565                     const struct GNUNET_PeerIdentity *peers)
1566 {
1567   const struct RPSPeer *rps_peer = cls;
1568
1569   for (uint64_t i = 0; i < num_peers; i++)
1570   {
1571     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1572                 "Peer %" PRIu32 " received [%s] from stream.\n",
1573                 rps_peer->index,
1574                 GNUNET_i2s (&peers[i]));
1575     if (0 != rps_peer->index &&
1576         0 == memcmp (&peers[i],
1577                      &rps_peers[0].peer_id,
1578                      sizeof (struct GNUNET_PeerIdentity)))
1579     {
1580       GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "Received a peer id outside sub\n");
1581       ok = 1;
1582     }
1583     else if (0 == rps_peer->index &&
1584              0 != memcmp (&peers[i],
1585                           &rps_peers[0].peer_id,
1586                           sizeof (struct GNUNET_PeerIdentity)))
1587     {
1588       GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "Received a peer id outside sub (lonely)\n");
1589       ok = 1;
1590     }
1591   }
1592 }
1593
1594
1595 static void
1596 sub_post (struct RPSPeer *rps_peer)
1597 {
1598   if (0 != rps_peer->index) GNUNET_RPS_sub_stop (rps_peer->rps_handle, "test");
1599   else GNUNET_RPS_sub_stop (rps_peer->rps_handle, "lonely");
1600 }
1601
1602
1603 static void
1604 sub_pre (struct RPSPeer *rps_peer, struct GNUNET_RPS_Handle *h)
1605 {
1606   (void) rps_peer;
1607
1608   if (0 != rps_peer->index) GNUNET_RPS_sub_start (h, "test");
1609   else GNUNET_RPS_sub_start (h, "lonely"); /* have a group of one */
1610   rps_peer->rps_srh = GNUNET_RPS_stream_request (h,
1611                                                  &got_stream_peer_cb,
1612                                                  rps_peer);
1613 }
1614
1615 /***********************************
1616  * PROFILER
1617 ***********************************/
1618
1619 /**
1620  * Callback to be called when RPS service is started or stopped at peers
1621  *
1622  * @param cls NULL
1623  * @param op the operation handle
1624  * @param emsg NULL on success; otherwise an error description
1625  */
1626 static void
1627 churn_cb (void *cls,
1628           struct GNUNET_TESTBED_Operation *op,
1629           const char *emsg)
1630 {
1631   (void) op;
1632   // FIXME
1633   struct OpListEntry *entry = cls;
1634
1635   if (GNUNET_YES == in_shutdown || GNUNET_YES == post_test)
1636   {
1637     return;
1638   }
1639
1640   GNUNET_TESTBED_operation_done (entry->op);
1641   if (NULL != emsg)
1642   {
1643     GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "Failed to start/stop RPS at a peer\n");
1644     GNUNET_SCHEDULER_shutdown ();
1645     return;
1646   }
1647   GNUNET_assert (0 != entry->delta);
1648
1649   num_peers_online += entry->delta;
1650
1651   if (PEER_GO_OFFLINE == entry->delta)
1652   { /* Peer hopefully just went offline */
1653     if (GNUNET_YES != rps_peers[entry->index].online)
1654     {
1655       GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
1656                   "peer %s was expected to go offline but is still marked as online\n",
1657                   GNUNET_i2s (rps_peers[entry->index].peer_id));
1658       GNUNET_break (0);
1659     }
1660     else
1661     {
1662       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1663                   "peer %s probably went offline as expected\n",
1664                   GNUNET_i2s (rps_peers[entry->index].peer_id));
1665     }
1666     rps_peers[entry->index].online = GNUNET_NO;
1667   }
1668
1669   else if (PEER_GO_ONLINE < entry->delta)
1670   { /* Peer hopefully just went online */
1671     if (GNUNET_NO != rps_peers[entry->index].online)
1672     {
1673       GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
1674                   "peer %s was expected to go online but is still marked as offline\n",
1675                   GNUNET_i2s (rps_peers[entry->index].peer_id));
1676       GNUNET_break (0);
1677     }
1678     else
1679     {
1680       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1681                   "peer %s probably went online as expected\n",
1682                   GNUNET_i2s (rps_peers[entry->index].peer_id));
1683       if (NULL != cur_test_run.pre_test)
1684       {
1685         cur_test_run.pre_test (&rps_peers[entry->index],
1686             rps_peers[entry->index].rps_handle);
1687         schedule_missing_requests (&rps_peers[entry->index]);
1688       }
1689     }
1690     rps_peers[entry->index].online = GNUNET_YES;
1691   }
1692   else
1693   {
1694     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
1695         "Invalid value for delta: %i\n", entry->delta);
1696     GNUNET_break (0);
1697   }
1698
1699   GNUNET_CONTAINER_DLL_remove (oplist_head, oplist_tail, entry);
1700   rps_peers[entry->index].entry_op_manage = NULL;
1701   GNUNET_free (entry);
1702   //if (num_peers_in_round[current_round] == peers_running)
1703   //  run_round ();
1704 }
1705
1706 /**
1707  * @brief Set the rps-service up or down for a specific peer
1708  *
1709  * @param i index of action
1710  * @param j index of peer
1711  * @param delta (#PEER_ONLINE_DELTA) down (-1) or up (1)
1712  * @param prob_go_on_off the probability of the action
1713  */
1714 static void
1715 manage_service_wrapper (unsigned int i, unsigned int j,
1716                         enum PEER_ONLINE_DELTA delta,
1717                         double prob_go_on_off)
1718 {
1719   struct OpListEntry *entry = NULL;
1720   uint32_t prob;
1721
1722   /* make sure that management operation is not already scheduled */
1723   if (NULL != rps_peers[j].entry_op_manage)
1724   {
1725     return;
1726   }
1727
1728   prob = GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
1729                                    UINT32_MAX);
1730   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1731               "%u. selected peer (%u: %s) is %s.\n",
1732               i,
1733               j,
1734               GNUNET_i2s (rps_peers[j].peer_id),
1735               (PEER_GO_ONLINE == delta) ? "online" : "offline");
1736   if (prob < prob_go_on_off * UINT32_MAX)
1737   {
1738     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1739                 "%s goes %s\n",
1740                 GNUNET_i2s (rps_peers[j].peer_id),
1741                 (PEER_GO_OFFLINE == delta) ? "offline" : "online");
1742
1743     if (PEER_GO_OFFLINE == delta)
1744       cancel_pending_req_rep (&rps_peers[j]);
1745     entry = make_oplist_entry ();
1746     entry->delta = delta;
1747     entry->index = j;
1748     entry->op = GNUNET_TESTBED_peer_manage_service (NULL,
1749                                                     testbed_peers[j],
1750                                                     "rps",
1751                                                     &churn_cb,
1752                                                     entry,
1753                                                     (PEER_GO_OFFLINE == delta) ? 0 : 1);
1754     rps_peers[j].entry_op_manage = entry;
1755   }
1756 }
1757
1758
1759 static void
1760 churn (void *cls)
1761 {
1762   (void) cls;
1763   unsigned int i;
1764   unsigned int j;
1765   double portion_online;
1766   unsigned int *permut;
1767   double prob_go_offline;
1768   double portion_go_online;
1769   double portion_go_offline;
1770
1771   if (GNUNET_YES == in_shutdown || GNUNET_YES == post_test)
1772   {
1773     return;
1774   }
1775   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1776               "Churn function executing\n");
1777
1778   churn_task = NULL; /* Should be invalid by now */
1779
1780   /* Compute the probability for an online peer to go offline
1781    * this round */
1782   portion_online = num_peers_online * 1.0 / num_peers;
1783   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1784               "Portion online: %f\n",
1785               portion_online);
1786   portion_go_online = ((1 - portion_online) * .5 * .66);
1787   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1788               "Portion that should go online: %f\n",
1789               portion_go_online);
1790   portion_go_offline = (portion_online + portion_go_online) - .75;
1791   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1792               "Portion that probably goes offline: %f\n",
1793               portion_go_offline);
1794   prob_go_offline = portion_go_offline / (portion_online * .5);
1795   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1796               "Probability of a selected online peer to go offline: %f\n",
1797               prob_go_offline);
1798
1799   permut = GNUNET_CRYPTO_random_permute (GNUNET_CRYPTO_QUALITY_WEAK,
1800                                          (unsigned int) num_peers);
1801
1802   /* Go over 50% randomly chosen peers */
1803   for (i = 0; i < .5 * num_peers; i++)
1804   {
1805     j = permut[i];
1806
1807     /* If online, shut down with certain probability */
1808     if (GNUNET_YES == rps_peers[j].online)
1809     {
1810       manage_service_wrapper (i, j, -1, prob_go_offline);
1811     }
1812
1813     /* If offline, restart with certain probability */
1814     else if (GNUNET_NO == rps_peers[j].online)
1815     {
1816       manage_service_wrapper (i, j, 1, 0.66);
1817     }
1818   }
1819
1820   GNUNET_free (permut);
1821
1822   churn_task = GNUNET_SCHEDULER_add_delayed (
1823         GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 2),
1824         churn,
1825         NULL);
1826 }
1827
1828
1829 /**
1830  * Initialise given RPSPeer
1831  */
1832 static void profiler_init_peer (struct RPSPeer *rps_peer)
1833 {
1834   if (num_peers - 1 == rps_peer->index)
1835     rps_peer->num_ids_to_request = cur_test_run.num_requests;
1836 }
1837
1838
1839 /**
1840  * Callback to call on receipt of a reply
1841  *
1842  * @param cls closure
1843  * @param n number of peers
1844  * @param recv_peers the received peers
1845  */
1846 static void
1847 profiler_reply_handle (void *cls,
1848                       uint64_t n,
1849                       const struct GNUNET_PeerIdentity *recv_peers)
1850 {
1851   struct RPSPeer *rps_peer;
1852   struct RPSPeer *rcv_rps_peer;
1853   char *file_name;
1854   char *file_name_dh;
1855   unsigned int i;
1856   struct PendingReply *pending_rep = (struct PendingReply *) cls;
1857
1858   rps_peer = pending_rep->rps_peer;
1859   file_name = "/tmp/rps/received_ids";
1860   file_name_dh = "/tmp/rps/diehard_input";
1861   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1862               "[%s] got %" PRIu64 " peers:\n",
1863               GNUNET_i2s (rps_peer->peer_id),
1864               n);
1865   for (i = 0; i < n; i++)
1866   {
1867     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1868                 "%u: %s\n",
1869                 i,
1870                 GNUNET_i2s (&recv_peers[i]));
1871     tofile (file_name,
1872              "%s\n",
1873              GNUNET_i2s_full (&recv_peers[i]));
1874     rcv_rps_peer = GNUNET_CONTAINER_multipeermap_get (peer_map, &recv_peers[i]);
1875     GNUNET_assert (NULL != rcv_rps_peer);
1876     tofile (file_name_dh,
1877              "%" PRIu32 "\n",
1878              (uint32_t) rcv_rps_peer->index);
1879   }
1880   default_reply_handle (cls, n, recv_peers);
1881 }
1882
1883
1884 static void
1885 profiler_cb (struct RPSPeer *rps_peer)
1886 {
1887   if (GNUNET_YES == in_shutdown || GNUNET_YES == post_test)
1888   {
1889     return;
1890   }
1891
1892   /* Start churn */
1893   if (HAVE_CHURN == cur_test_run.have_churn && NULL == churn_task)
1894   {
1895     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1896                 "Starting churn task\n");
1897     churn_task = GNUNET_SCHEDULER_add_delayed (
1898           GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 5),
1899           churn,
1900           NULL);
1901   } else {
1902     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1903                 "Not starting churn task\n");
1904   }
1905
1906   /* Only request peer ids at one peer.
1907    * (It's the before-last because last one is target of the focussed attack.)
1908    */
1909   if (eval_peer == rps_peer)
1910     schedule_missing_requests (rps_peer);
1911 }
1912
1913 /**
1914  * Function called from #profiler_eval with a filename.
1915  *
1916  * @param cls closure
1917  * @param filename complete filename (absolute path)
1918  * @return #GNUNET_OK to continue to iterate,
1919  *  #GNUNET_NO to stop iteration with no error,
1920  *  #GNUNET_SYSERR to abort iteration with error!
1921  */
1922 int
1923 file_name_cb (void *cls, const char *filename)
1924 {
1925   (void) cls;
1926
1927   if (NULL != strstr (filename, "sampler_el"))
1928   {
1929     struct RPS_SamplerElement *s_elem;
1930     struct GNUNET_CRYPTO_AuthKey auth_key;
1931     const char *key_char;
1932     uint32_t i;
1933
1934     key_char = filename + 20; /* Length of "/tmp/rps/sampler_el-" */
1935     tofile (filename, "--------------------------\n");
1936
1937     auth_key = string_to_auth_key (key_char);
1938     s_elem = RPS_sampler_elem_create ();
1939     RPS_sampler_elem_set (s_elem, auth_key);
1940
1941     for (i = 0; i < num_peers; i++)
1942     {
1943       RPS_sampler_elem_next (s_elem, &rps_peer_ids[i]);
1944     }
1945     RPS_sampler_elem_destroy (s_elem);
1946   }
1947   return GNUNET_OK;
1948 }
1949
1950 /**
1951  * This is run after the test finished.
1952  *
1953  * Compute all perfect samples.
1954  */
1955 int
1956 profiler_eval (void)
1957 {
1958   /* Compute perfect sample for each sampler element */
1959   if (-1 == GNUNET_DISK_directory_scan ("/tmp/rps/", file_name_cb, NULL))
1960   {
1961     GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "Scan of directory failed\n");
1962   }
1963
1964   return evaluate ();
1965 }
1966
1967
1968 /**
1969  * @brief is b in view of a?
1970  *
1971  * @param a
1972  * @param b
1973  *
1974  * @return
1975  */
1976 static int is_in_view (uint32_t a, uint32_t b)
1977 {
1978   uint32_t i;
1979   for (i = 0; i < rps_peers[a].cur_view_count; i++)
1980   {
1981     if (0 == memcmp (rps_peers[b].peer_id,
1982           &rps_peers[a].cur_view[i],
1983           sizeof (struct GNUNET_PeerIdentity)))
1984     {
1985       return GNUNET_YES;
1986     }
1987   }
1988   return GNUNET_NO;
1989 }
1990
1991 static uint32_t get_idx_of_pid (const struct GNUNET_PeerIdentity *pid)
1992 {
1993   uint32_t i;
1994
1995   for (i = 0; i < num_peers; i++)
1996   {
1997     if (0 == memcmp (pid,
1998           rps_peers[i].peer_id,
1999           sizeof (struct GNUNET_PeerIdentity)))
2000     {
2001       return i;
2002     }
2003   }
2004   //return 0; /* Should not happen - make compiler happy */
2005   GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
2006              "No known _PeerIdentity %s!\n",
2007              GNUNET_i2s_full (pid));
2008   GNUNET_assert (0);
2009 }
2010
2011 /**
2012  * @brief Counts number of peers in view of a that have b in their view
2013  *
2014  * @param a
2015  * @param uint32_tb
2016  *
2017  * @return
2018  */
2019 static uint32_t count_containing_views (uint32_t a, uint32_t b)
2020 {
2021   uint32_t i;
2022   uint32_t peer_idx;
2023   uint32_t count = 0;
2024
2025   for (i = 0; i < rps_peers[a].cur_view_count; i++)
2026   {
2027     peer_idx = get_idx_of_pid (&rps_peers[a].cur_view[i]);
2028     if (GNUNET_YES == is_in_view (peer_idx, b))
2029     {
2030       count++;
2031     }
2032   }
2033   return count;
2034 }
2035
2036 /**
2037  * @brief Computes the probability for each other peer to be selected by the
2038  * sampling process based on the views of all peers
2039  *
2040  * @param peer_idx index of the peer that is about to sample
2041  */
2042 static void compute_probabilities (uint32_t peer_idx)
2043 {
2044   //double probs[num_peers] = { 0 };
2045   double probs[num_peers];
2046   size_t probs_as_str_size = (num_peers * 10 + 1) * sizeof (char);
2047   char *probs_as_str = GNUNET_malloc (probs_as_str_size);
2048   char *probs_as_str_cpy;
2049   uint32_t i;
2050   double prob_push;
2051   double prob_pull;
2052   uint32_t view_size;
2053   uint32_t cont_views;
2054   uint32_t number_of_being_in_pull_events;
2055   int tmp;
2056   uint32_t count_non_zero_prob = 0;
2057
2058   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2059       "Computing probabilities for peer %" PRIu32 "\n", peer_idx);
2060   /* Firstly without knowledge of old views */
2061   for (i = 0; i < num_peers; i++)
2062   {
2063     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2064         "\tfor peer %" PRIu32 ":\n", i);
2065     view_size = rps_peers[i].cur_view_count;
2066     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2067         "\t\tview_size: %" PRIu32 "\n", view_size);
2068     /* For peer i the probability of being sampled is
2069      * evenly distributed among all possibly observed peers. */
2070     /* We could have observed a peer in three cases:
2071      *   1. peer sent a push
2072      *   2. peer was contained in a pull reply
2073      *   3. peer was in history (sampler) - ignored for now */
2074     /* 1. Probability of having received a push from peer i */
2075     if ((GNUNET_YES == is_in_view (i, peer_idx)) &&
2076         (1 <= (0.45 * view_size)))
2077     {
2078       prob_push = 1.0 * binom (0.45 * view_size, 1)
2079         /
2080         binom (view_size, 0.45 * view_size);
2081       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2082                  "\t\t%" PRIu32 " is in %" PRIu32 "'s view, prob: %f\n",
2083                  peer_idx,
2084                  i,
2085                  prob_push);
2086       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2087                  "\t\tposs choices from view: %" PRIu32 ", containing i: %" PRIu32 "\n",
2088                  binom (view_size, 0.45 * view_size),
2089                  binom (0.45 * view_size, 1));
2090     } else {
2091       prob_push = 0;
2092       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2093                  "\t\t%" PRIu32 " is not in %" PRIu32 "'s view, prob: 0\n",
2094                  peer_idx,
2095                  i);
2096     }
2097     /* 2. Probability of peer i being contained in pulls */
2098     view_size = rps_peers[peer_idx].cur_view_count;
2099     cont_views = count_containing_views (peer_idx, i);
2100     number_of_being_in_pull_events =
2101       (binom (view_size, 0.45 * view_size) -
2102        binom (view_size - cont_views, 0.45 * view_size));
2103     if (0 != number_of_being_in_pull_events)
2104     {
2105       prob_pull = number_of_being_in_pull_events
2106         /
2107         (1.0 * binom (view_size, 0.45 * view_size));
2108     } else
2109     {
2110       prob_pull = 0;
2111     }
2112     probs[i] = prob_push + prob_pull - (prob_push * prob_pull);
2113     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2114                "\t\t%" PRIu32 " has %" PRIu32 " of %" PRIu32
2115                " peers in its view who know %" PRIu32 " prob: %f\n",
2116                peer_idx,
2117                cont_views,
2118                view_size,
2119                i,
2120                prob_pull);
2121     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2122                "\t\tnumber of possible pull combinations: %" PRIu32 "\n",
2123                binom (view_size, 0.45 * view_size));
2124     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2125                "\t\tnumber of possible pull combinations without %" PRIu32
2126                ": %" PRIu32 "\n",
2127                i,
2128                binom (view_size - cont_views, 0.45 * view_size));
2129     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2130                "\t\tnumber of possible pull combinations with %" PRIu32
2131                ": %" PRIu32 "\n",
2132                i,
2133                number_of_being_in_pull_events);
2134
2135     if (0 != probs[i]) count_non_zero_prob++;
2136   }
2137   /* normalize */
2138   if (0 != count_non_zero_prob)
2139   {
2140     for (i = 0; i < num_peers; i++)
2141     {
2142       probs[i] = probs[i] * (1.0 / count_non_zero_prob);
2143     }
2144   } else {
2145     for (i = 0; i < num_peers; i++)
2146     {
2147       probs[i] = 0;
2148     }
2149   }
2150   /* str repr */
2151   for (i = 0; i < num_peers; i++)
2152   {
2153     probs_as_str_cpy = GNUNET_strndup (probs_as_str, probs_as_str_size);
2154     tmp = GNUNET_snprintf (probs_as_str,
2155                            probs_as_str_size,
2156                            "%s %7.6f", probs_as_str_cpy, probs[i]);
2157     GNUNET_free (probs_as_str_cpy);
2158     GNUNET_assert (0 <= tmp);
2159   }
2160
2161   to_file_w_len (rps_peers[peer_idx].file_name_probs,
2162                  probs_as_str_size,
2163                  probs_as_str);
2164   GNUNET_free (probs_as_str);
2165 }
2166
2167 /**
2168  * @brief This counts the number of peers in which views a given peer occurs.
2169  *
2170  * It also stores this value in the rps peer.
2171  *
2172  * @param peer_idx the index of the peer to count the representation
2173  *
2174  * @return the number of occurrences
2175  */
2176 static uint32_t count_peer_in_views_2 (uint32_t peer_idx)
2177 {
2178   uint32_t i, j;
2179   uint32_t count = 0;
2180
2181   for (i = 0; i < num_peers; i++) /* Peer in which view is counted */
2182   {
2183     for (j = 0; j < rps_peers[i].cur_view_count; j++) /* entry in view */
2184     {
2185       if (0 == memcmp (rps_peers[peer_idx].peer_id,
2186             &rps_peers[i].cur_view[j],
2187             sizeof (struct GNUNET_PeerIdentity)))
2188       {
2189         count++;
2190         break;
2191       }
2192     }
2193   }
2194   rps_peers[peer_idx].count_in_views = count;
2195   return count;
2196 }
2197
2198 static uint32_t cumulated_view_sizes ()
2199 {
2200   uint32_t i;
2201
2202   view_sizes = 0;
2203   for (i = 0; i < num_peers; i++) /* Peer in which view is counted */
2204   {
2205     view_sizes += rps_peers[i].cur_view_count;
2206   }
2207   return view_sizes;
2208 }
2209
2210 static void count_peer_in_views (uint32_t *count_peers)
2211 {
2212   uint32_t i, j;
2213
2214   for (i = 0; i < num_peers; i++) /* Peer in which view is counted */
2215   {
2216     for (j = 0; j < rps_peers[i].cur_view_count; j++) /* entry in view */
2217     {
2218       if (0 == memcmp (rps_peers[i].peer_id,
2219             &rps_peers[i].cur_view[j],
2220             sizeof (struct GNUNET_PeerIdentity)))
2221       {
2222         count_peers[i]++;
2223       }
2224     }
2225   }
2226 }
2227
2228 void compute_diversity ()
2229 {
2230   uint32_t i;
2231   /* ith entry represents the numer of occurrences in other peer's views */
2232   uint32_t *count_peers = GNUNET_new_array (num_peers, uint32_t);
2233   uint32_t views_total_size;
2234   double expected;
2235   /* deviation from expected number of peers */
2236   double *deviation = GNUNET_new_array (num_peers, double);
2237
2238   views_total_size = 0;
2239   expected = 0;
2240
2241   /* For each peer count its representation in other peer's views*/
2242   for (i = 0; i < num_peers; i++) /* Peer to count */
2243   {
2244     views_total_size += rps_peers[i].cur_view_count;
2245     count_peer_in_views (count_peers);
2246     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2247                "Counted representation of %" PRIu32 "th peer [%s]: %" PRIu32"\n",
2248                i,
2249                GNUNET_i2s (rps_peers[i].peer_id),
2250                count_peers[i]);
2251   }
2252
2253   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2254              "size of all views combined: %" PRIu32 "\n",
2255              views_total_size);
2256   expected = ((double) 1/num_peers) * views_total_size;
2257   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2258              "Expected number of occurrences of each peer in all views: %f\n",
2259              expected);
2260   for (i = 0; i < num_peers; i++) /* Peer to count */
2261   {
2262     deviation[i] = expected - count_peers[i];
2263     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2264                "Deviation from expectation: %f\n", deviation[i]);
2265   }
2266   GNUNET_free (count_peers);
2267   GNUNET_free (deviation);
2268 }
2269
2270 void print_view_sizes()
2271 {
2272   uint32_t i;
2273
2274   for (i = 0; i < num_peers; i++) /* Peer to count */
2275   {
2276     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2277                "View size of %" PRIu32 ". [%s] is %" PRIu32 "\n",
2278                i,
2279                GNUNET_i2s (rps_peers[i].peer_id),
2280                rps_peers[i].cur_view_count);
2281   }
2282 }
2283
2284 void all_views_updated_cb()
2285 {
2286   compute_diversity();
2287   print_view_sizes();
2288 }
2289
2290 void view_update_cb (void *cls,
2291                      uint64_t view_size,
2292                      const struct GNUNET_PeerIdentity *peers)
2293 {
2294   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2295               "View was updated (%" PRIu64 ")\n", view_size);
2296   struct RPSPeer *rps_peer = (struct RPSPeer *) cls;
2297   to_file ("/tmp/rps/view_sizes.txt",
2298          "%" PRIu64 " %" PRIu32 "",
2299          rps_peer->index,
2300          view_size);
2301   for (int i = 0; i < view_size; i++)
2302   {
2303     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2304                "\t%s\n", GNUNET_i2s (&peers[i]));
2305   }
2306   GNUNET_array_grow (rps_peer->cur_view,
2307                      rps_peer->cur_view_count,
2308                      view_size);
2309   //*rps_peer->cur_view = *peers;
2310   GNUNET_memcpy (rps_peer->cur_view,
2311                  peers,
2312                  view_size * sizeof (struct GNUNET_PeerIdentity));
2313   to_file ("/tmp/rps/count_in_views.txt",
2314          "%" PRIu64 " %" PRIu32 "",
2315          rps_peer->index,
2316          count_peer_in_views_2 (rps_peer->index));
2317   cumulated_view_sizes();
2318   if (0 != view_size)
2319   {
2320     to_file ("/tmp/rps/repr.txt",
2321            "%" PRIu64 /* index */
2322            " %" PRIu32 /* occurrence in views */
2323            " %" PRIu32 /* view sizes */
2324            " %f" /* fraction of repr in views */
2325            " %f" /* average view size */
2326            " %f" /* prob of occurrence in view slot */
2327            " %f" "", /* exp frac of repr in views */
2328            rps_peer->index,
2329            count_peer_in_views_2 (rps_peer->index),
2330            view_sizes,
2331            count_peer_in_views_2 (rps_peer->index) / (view_size * 1.0), /* fraction of representation in views */
2332            view_sizes / (view_size * 1.0), /* average view size */
2333            1.0 /view_size, /* prob of occurrence in view slot */
2334            (1.0/view_size) * (view_sizes/view_size) /* expected fraction of repr in views */
2335            );
2336   }
2337   compute_probabilities (rps_peer->index);
2338   all_views_updated_cb();
2339 }
2340
2341 static void
2342 pre_profiler (struct RPSPeer *rps_peer, struct GNUNET_RPS_Handle *h)
2343 {
2344   rps_peer->file_name_probs =
2345     store_prefix_file_name (rps_peer->peer_id, "probs");
2346   GNUNET_RPS_view_request (h, 0, view_update_cb, rps_peer);
2347 }
2348
2349 void write_final_stats (void){
2350   uint32_t i;
2351
2352   for (i = 0; i < num_peers; i++)
2353   {
2354     to_file ("/tmp/rps/final_stats.dat",
2355              "%" PRIu32 " " /* index */
2356              "%s %" /* id */
2357              PRIu64 " %" /* rounds */
2358              PRIu64 " %" PRIu64 " %" PRIu64 " %" PRIu64 " %" PRIu64 " %" PRIu64 " %" /* blocking */
2359              PRIu64 " %" PRIu64 " %" PRIu64 " %" /* issued */
2360              PRIu64 " %" PRIu64 " %" PRIu64 " %" /* sent */
2361              PRIu64 " %" PRIu64 " %" PRIu64 /* recv */,
2362              i,
2363              GNUNET_i2s (rps_peers[i].peer_id),
2364              rps_peers[i].num_rounds,
2365              rps_peers[i].num_blocks,
2366              rps_peers[i].num_blocks_many_push,
2367              rps_peers[i].num_blocks_no_push,
2368              rps_peers[i].num_blocks_no_pull,
2369              rps_peers[i].num_blocks_many_push_no_pull,
2370              rps_peers[i].num_blocks_no_push_no_pull,
2371              rps_peers[i].num_issued_push,
2372              rps_peers[i].num_issued_pull_req,
2373              rps_peers[i].num_issued_pull_rep,
2374              rps_peers[i].num_sent_push,
2375              rps_peers[i].num_sent_pull_req,
2376              rps_peers[i].num_sent_pull_rep,
2377              rps_peers[i].num_recv_push,
2378              rps_peers[i].num_recv_pull_req,
2379              rps_peers[i].num_recv_pull_rep);
2380   }
2381 }
2382
2383 /**
2384  * Continuation called by #GNUNET_STATISTICS_get() functions.
2385  *
2386  * Remembers that this specific statistics value was received for this peer.
2387  * Checks whether all peers received their statistics yet.
2388  * Issues the shutdown.
2389  *
2390  * @param cls closure
2391  * @param success #GNUNET_OK if statistics were
2392  *        successfully obtained, #GNUNET_SYSERR if not.
2393  */
2394 void
2395 post_test_shutdown_ready_cb (void *cls,
2396                              int success)
2397 {
2398   struct STATcls *stat_cls = (struct STATcls *) cls;
2399   struct RPSPeer *rps_peer = stat_cls->rps_peer;
2400   if (GNUNET_OK == success)
2401   {
2402     /* set flag that we we got the value */
2403     rps_peer->stat_collected_flags |= stat_cls->stat_type;
2404   } else {
2405     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
2406         "Peer %u did not receive statistics value\n",
2407         rps_peer->index);
2408     GNUNET_free (stat_cls);
2409     GNUNET_break (0);
2410   }
2411
2412   if (NULL != rps_peer->stat_op &&
2413       GNUNET_YES == check_statistics_collect_completed_single_peer (rps_peer))
2414   {
2415     GNUNET_TESTBED_operation_done (rps_peer->stat_op);
2416   }
2417
2418   write_final_stats ();
2419   if (GNUNET_YES == check_statistics_collect_completed())
2420   {
2421     //write_final_stats ();
2422     GNUNET_free (stat_cls);
2423     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2424         "Shutting down\n");
2425     GNUNET_SCHEDULER_shutdown ();
2426   } else {
2427     GNUNET_free (stat_cls);
2428   }
2429 }
2430
2431 /**
2432  * @brief Converts string representation to the corresponding #STAT_TYPE enum.
2433  *
2434  * @param stat_str string representation of statistics specifier
2435  *
2436  * @return corresponding enum
2437  */
2438 enum STAT_TYPE stat_str_2_type (const char *stat_str)
2439 {
2440   if (0 == strncmp ("# rounds blocked - no pull replies", stat_str, strlen ("# rounds blocked - no pull replies")))
2441   {
2442     return STAT_TYPE_BLOCKS_NO_PULL;
2443   }
2444   else if (0 == strncmp ("# rounds blocked - too many pushes, no pull replies", stat_str, strlen ("# rounds blocked - too many pushes, no pull replies")))
2445   {
2446     return STAT_TYPE_BLOCKS_MANY_PUSH_NO_PULL;
2447   }
2448   else if (0 == strncmp ("# rounds blocked - too many pushes", stat_str, strlen ("# rounds blocked - too many pushes")))
2449   {
2450     return STAT_TYPE_BLOCKS_MANY_PUSH;
2451   }
2452   else if (0 == strncmp ("# rounds blocked - no pushes, no pull replies", stat_str, strlen ("# rounds blocked - no pushes, no pull replies")))
2453   {
2454     return STAT_TYPE_BLOCKS_NO_PUSH_NO_PULL;
2455   }
2456   else if (0 == strncmp ("# rounds blocked - no pushes", stat_str, strlen ("# rounds blocked - no pushes")))
2457   {
2458     return STAT_TYPE_BLOCKS_NO_PUSH;
2459   }
2460   else if (0 == strncmp ("# rounds blocked", stat_str, strlen ("# rounds blocked")))
2461   {
2462     return STAT_TYPE_BLOCKS;
2463   }
2464   else if (0 == strncmp ("# rounds", stat_str, strlen ("# rounds")))
2465   {
2466     return STAT_TYPE_ROUNDS;
2467   }
2468   else if (0 == strncmp ("# push send issued", stat_str, strlen ("# push send issued")))
2469   {
2470     return STAT_TYPE_ISSUED_PUSH_SEND;
2471   }
2472   else if (0 == strncmp ("# pull request send issued", stat_str, strlen ("# pull request send issued")))
2473   {
2474     return STAT_TYPE_ISSUED_PULL_REQ;
2475   }
2476   else if (0 == strncmp ("# pull reply send issued", stat_str, strlen ("# pull reply send issued")))
2477   {
2478     return STAT_TYPE_ISSUED_PULL_REP;
2479   }
2480   else if (0 == strncmp ("# pushes sent", stat_str, strlen ("# pushes sent")))
2481   {
2482     return STAT_TYPE_SENT_PUSH_SEND;
2483   }
2484   else if (0 == strncmp ("# pull requests sent", stat_str, strlen ("# pull requests sent")))
2485   {
2486     return STAT_TYPE_SENT_PULL_REQ;
2487   }
2488   else if (0 == strncmp ("# pull replys sent", stat_str, strlen ("# pull replys sent")))
2489   {
2490     return STAT_TYPE_SENT_PULL_REP;
2491   }
2492   else if (0 == strncmp ("# push message received", stat_str, strlen ("# push message received")))
2493   {
2494     return STAT_TYPE_RECV_PUSH_SEND;
2495   }
2496   else if (0 == strncmp ("# pull request message received", stat_str, strlen ("# pull request message received")))
2497   {
2498     return STAT_TYPE_RECV_PULL_REQ;
2499   }
2500   else if (0 == strncmp ("# pull reply messages received", stat_str, strlen ("# pull reply messages received")))
2501   {
2502     return STAT_TYPE_RECV_PULL_REP;
2503   }
2504   return STAT_TYPE_MAX;
2505 }
2506
2507
2508 /**
2509  * @brief Converts #STAT_TYPE enum to the equivalent string representation that
2510  * is stored with the statistics service.
2511  *
2512  * @param stat_type #STAT_TYPE enum
2513  *
2514  * @return string representation that matches statistics value
2515  */
2516 char* stat_type_2_str (enum STAT_TYPE stat_type)
2517 {
2518   switch (stat_type)
2519   {
2520     case STAT_TYPE_ROUNDS:
2521       return "# rounds";
2522     case STAT_TYPE_BLOCKS:
2523       return "# rounds blocked";
2524     case STAT_TYPE_BLOCKS_MANY_PUSH:
2525       return "# rounds blocked - too many pushes";
2526     case STAT_TYPE_BLOCKS_NO_PUSH:
2527       return "# rounds blocked - no pushes";
2528     case STAT_TYPE_BLOCKS_NO_PULL:
2529       return "# rounds blocked - no pull replies";
2530     case STAT_TYPE_BLOCKS_MANY_PUSH_NO_PULL:
2531       return "# rounds blocked - too many pushes, no pull replies";
2532     case STAT_TYPE_BLOCKS_NO_PUSH_NO_PULL:
2533       return "# rounds blocked - no pushes, no pull replies";
2534     case STAT_TYPE_ISSUED_PUSH_SEND:
2535       return "# push send issued";
2536     case STAT_TYPE_ISSUED_PULL_REQ:
2537       return "# pull request send issued";
2538     case STAT_TYPE_ISSUED_PULL_REP:
2539       return "# pull reply send issued";
2540     case STAT_TYPE_SENT_PUSH_SEND:
2541       return "# pushes sent";
2542     case STAT_TYPE_SENT_PULL_REQ:
2543       return "# pull requests sent";
2544     case STAT_TYPE_SENT_PULL_REP:
2545       return "# pull replys sent";
2546     case STAT_TYPE_RECV_PUSH_SEND:
2547       return "# push message received";
2548     case STAT_TYPE_RECV_PULL_REQ:
2549       return "# pull request message received";
2550     case STAT_TYPE_RECV_PULL_REP:
2551       return "# pull reply messages received";
2552     case STAT_TYPE_MAX:
2553     default:
2554       return "ERROR";
2555       ;
2556   }
2557 }
2558
2559 /**
2560  * Callback function to process statistic values.
2561  *
2562  * @param cls closure
2563  * @param subsystem name of subsystem that created the statistic
2564  * @param name the name of the datum
2565  * @param value the current value
2566  * @param is_persistent #GNUNET_YES if the value is persistent, #GNUNET_NO if not
2567  * @return #GNUNET_OK to continue, #GNUNET_SYSERR to abort iteration
2568  */
2569 int
2570 stat_iterator (void *cls,
2571                const char *subsystem,
2572                const char *name,
2573                uint64_t value,
2574                int is_persistent)
2575 {
2576   (void) subsystem;
2577   (void) is_persistent;
2578   const struct STATcls *stat_cls = (const struct STATcls *) cls;
2579   struct RPSPeer *rps_peer = (struct RPSPeer *) stat_cls->rps_peer;
2580   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Got stat value: %s - %" PRIu64 "\n",
2581       //stat_type_2_str (stat_cls->stat_type),
2582       name,
2583       value);
2584   to_file (rps_peer->file_name_stats,
2585           "%s: %" PRIu64 "\n",
2586           name,
2587           value);
2588   switch (stat_str_2_type (name))
2589   {
2590     case STAT_TYPE_ROUNDS:
2591       rps_peer->num_rounds = value;
2592       break;
2593     case STAT_TYPE_BLOCKS:
2594       rps_peer->num_blocks = value;
2595       break;
2596     case STAT_TYPE_BLOCKS_MANY_PUSH:
2597       rps_peer->num_blocks_many_push = value;
2598       break;
2599     case STAT_TYPE_BLOCKS_NO_PUSH:
2600       rps_peer->num_blocks_no_push = value;
2601       break;
2602     case STAT_TYPE_BLOCKS_NO_PULL:
2603       rps_peer->num_blocks_no_pull = value;
2604       break;
2605     case STAT_TYPE_BLOCKS_MANY_PUSH_NO_PULL:
2606       rps_peer->num_blocks_many_push_no_pull = value;
2607       break;
2608     case STAT_TYPE_BLOCKS_NO_PUSH_NO_PULL:
2609       rps_peer->num_blocks_no_push_no_pull = value;
2610       break;
2611     case STAT_TYPE_ISSUED_PUSH_SEND:
2612       rps_peer->num_issued_push = value;
2613       break;
2614     case STAT_TYPE_ISSUED_PULL_REQ:
2615       rps_peer->num_issued_pull_req = value;
2616       break;
2617     case STAT_TYPE_ISSUED_PULL_REP:
2618       rps_peer->num_issued_pull_rep = value;
2619       break;
2620     case STAT_TYPE_SENT_PUSH_SEND:
2621       rps_peer->num_sent_push = value;
2622       break;
2623     case STAT_TYPE_SENT_PULL_REQ:
2624       rps_peer->num_sent_pull_req = value;
2625       break;
2626     case STAT_TYPE_SENT_PULL_REP:
2627       rps_peer->num_sent_pull_rep = value;
2628       break;
2629     case STAT_TYPE_RECV_PUSH_SEND:
2630       rps_peer->num_recv_push = value;
2631       break;
2632     case STAT_TYPE_RECV_PULL_REQ:
2633       rps_peer->num_recv_pull_req = value;
2634       break;
2635     case STAT_TYPE_RECV_PULL_REP:
2636       rps_peer->num_recv_pull_rep = value;
2637       break;
2638     case STAT_TYPE_MAX:
2639     default:
2640       GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
2641                  "Unknown statistics string: %s\n",
2642                  name);
2643       break;
2644   }
2645   return GNUNET_OK;
2646 }
2647
2648 void post_profiler (struct RPSPeer *rps_peer)
2649 {
2650   if (COLLECT_STATISTICS != cur_test_run.have_collect_statistics)
2651   {
2652     return;
2653   }
2654
2655   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2656       "Going to request statistic values with mask 0x%" PRIx32 "\n",
2657       cur_test_run.stat_collect_flags);
2658
2659   struct STATcls *stat_cls;
2660   uint32_t stat_type;
2661   for (stat_type = STAT_TYPE_ROUNDS;
2662       stat_type < STAT_TYPE_MAX;
2663       stat_type = stat_type <<1)
2664   {
2665     if (stat_type & cur_test_run.stat_collect_flags)
2666     {
2667       stat_cls = GNUNET_malloc (sizeof (struct STATcls));
2668       stat_cls->rps_peer = rps_peer;
2669       stat_cls->stat_type = stat_type;
2670       rps_peer->file_name_stats =
2671         store_prefix_file_name (rps_peer->peer_id, "stats");
2672       GNUNET_STATISTICS_get (rps_peer->stats_h,
2673                              "rps",
2674                              stat_type_2_str (stat_type),
2675                              post_test_shutdown_ready_cb,
2676                              stat_iterator,
2677                              (struct STATcls *) stat_cls);
2678       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2679           "Requested statistics for %s (peer %" PRIu32 ")\n",
2680           stat_type_2_str (stat_type),
2681           rps_peer->index);
2682     }
2683   }
2684 }
2685
2686
2687 /***********************************************************************
2688  * /Definition of tests
2689 ***********************************************************************/
2690
2691
2692 /**
2693  * Actual "main" function for the testcase.
2694  *
2695  * @param cls closure
2696  * @param h the run handle
2697  * @param n_peers number of peers in 'peers'
2698  * @param peers handle to peers run in the testbed
2699  * @param links_succeeded the number of overlay link connection attempts that
2700  *          succeeded
2701  * @param links_failed the number of overlay link connection attempts that
2702  *          failed
2703  */
2704 static void
2705 run (void *cls,
2706      struct GNUNET_TESTBED_RunHandle *h,
2707      unsigned int n_peers,
2708      struct GNUNET_TESTBED_Peer **peers,
2709      unsigned int links_succeeded,
2710      unsigned int links_failed)
2711 {
2712   (void) cls;
2713   (void) h;
2714   (void) links_failed;
2715   unsigned int i;
2716   struct OpListEntry *entry;
2717
2718   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "RUN was called\n");
2719
2720   /* Check whether we timed out */
2721   if (n_peers != num_peers ||
2722       NULL == peers ||
2723       0 == links_succeeded)
2724   {
2725     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Going down due to args (eg. timeout)\n");
2726     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "\tn_peers: %u\n", n_peers);
2727     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "\tnum_peers: %" PRIu32 "\n", num_peers);
2728     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "\tpeers: %p\n", peers);
2729     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "\tlinks_succeeded: %u\n", links_succeeded);
2730     GNUNET_SCHEDULER_shutdown ();
2731     return;
2732   }
2733
2734
2735   /* Initialize peers */
2736   testbed_peers = peers;
2737   num_peers_online = 0;
2738   for (i = 0; i < num_peers; i++)
2739   {
2740     entry = make_oplist_entry ();
2741     entry->index = i;
2742     rps_peers[i].index = i;
2743     if (NULL != cur_test_run.init_peer)
2744       cur_test_run.init_peer (&rps_peers[i]);
2745     if (NO_COLLECT_VIEW == cur_test_run.have_collect_view)
2746     {
2747       rps_peers->cur_view_count = 0;
2748       rps_peers->cur_view = NULL;
2749     }
2750     entry->op = GNUNET_TESTBED_peer_get_information (peers[i],
2751                                                      GNUNET_TESTBED_PIT_IDENTITY,
2752                                                      &info_cb,
2753                                                      entry);
2754   }
2755
2756   /* Bring peers up */
2757   GNUNET_assert (num_peers == n_peers);
2758   for (i = 0; i < n_peers; i++)
2759   {
2760     rps_peers[i].index = i;
2761     rps_peers[i].op =
2762       GNUNET_TESTBED_service_connect (&rps_peers[i],
2763                                       peers[i],
2764                                       "rps",
2765                                       &rps_connect_complete_cb,
2766                                       &rps_peers[i],
2767                                       &rps_connect_adapter,
2768                                       &rps_disconnect_adapter,
2769                                       &rps_peers[i]);
2770     /* Connect all peers to statistics service */
2771     if (COLLECT_STATISTICS == cur_test_run.have_collect_statistics)
2772     {
2773       rps_peers[i].stat_op =
2774         GNUNET_TESTBED_service_connect (NULL,
2775                                         peers[i],
2776                                         "statistics",
2777                                         stat_complete_cb,
2778                                         &rps_peers[i],
2779                                         &stat_connect_adapter,
2780                                         &stat_disconnect_adapter,
2781                                         &rps_peers[i]);
2782     }
2783   }
2784
2785   if (NULL != churn_task)
2786     GNUNET_SCHEDULER_cancel (churn_task);
2787   post_test_task = GNUNET_SCHEDULER_add_delayed (timeout, &post_test_op, NULL);
2788   timeout = GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS,
2789       (timeout_s * 1.2) + 0.1 * num_peers);
2790   shutdown_task = GNUNET_SCHEDULER_add_delayed (timeout, &shutdown_op, NULL);
2791 }
2792
2793
2794 /**
2795  * Entry point for the testcase, sets up the testbed.
2796  *
2797  * @param argc unused
2798  * @param argv unused
2799  * @return 0 on success
2800  */
2801 int
2802 main (int argc, char *argv[])
2803 {
2804   int ret_value;
2805   (void) argc;
2806
2807   /* Defaults for tests */
2808   num_peers = 5;
2809   cur_test_run.name = "test-rps-default";
2810   cur_test_run.init_peer = default_init_peer;
2811   cur_test_run.pre_test = NULL;
2812   cur_test_run.reply_handle = default_reply_handle;
2813   cur_test_run.eval_cb = default_eval_cb;
2814   cur_test_run.post_test = NULL;
2815   cur_test_run.have_churn = HAVE_CHURN;
2816   cur_test_run.have_collect_statistics = NO_COLLECT_STATISTICS;
2817   cur_test_run.stat_collect_flags = 0;
2818   cur_test_run.have_collect_view = NO_COLLECT_VIEW;
2819   churn_task = NULL;
2820   timeout_s = 30;
2821
2822   if (strstr (argv[0], "malicious") != NULL)
2823   {
2824     cur_test_run.pre_test = mal_pre;
2825     cur_test_run.main_test = mal_cb;
2826     cur_test_run.init_peer = mal_init_peer;
2827     timeout_s = 40;
2828
2829     if (strstr (argv[0], "_1") != NULL)
2830     {
2831       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Test malicious peer type 1\n");
2832       cur_test_run.name = "test-rps-malicious_1";
2833       mal_type = 1;
2834     }
2835     else if (strstr (argv[0], "_2") != NULL)
2836     {
2837       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Test malicious peer type 2\n");
2838       cur_test_run.name = "test-rps-malicious_2";
2839       mal_type = 2;
2840     }
2841     else if (strstr (argv[0], "_3") != NULL)
2842     {
2843       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Test malicious peer type 3\n");
2844       cur_test_run.name = "test-rps-malicious_3";
2845       mal_type = 3;
2846     }
2847   }
2848
2849   else if (strstr (argv[0], "_single_req") != NULL)
2850   {
2851     GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "Test single request\n");
2852     cur_test_run.name = "test-rps-single-req";
2853     cur_test_run.main_test = single_req_cb;
2854     cur_test_run.have_churn = HAVE_NO_CHURN;
2855   }
2856
2857   else if (strstr (argv[0], "_delayed_reqs") != NULL)
2858   {
2859     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Test delayed requests\n");
2860     cur_test_run.name = "test-rps-delayed-reqs";
2861     cur_test_run.main_test = delay_req_cb;
2862     cur_test_run.have_churn = HAVE_NO_CHURN;
2863   }
2864
2865   else if (strstr (argv[0], "_seed_big") != NULL)
2866   {
2867     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Test seeding (num_peers > GNUNET_MAX_MESSAGE_SIZE)\n");
2868     num_peers = 1;
2869     cur_test_run.name = "test-rps-seed-big";
2870     cur_test_run.main_test = seed_big_cb;
2871     cur_test_run.eval_cb = no_eval;
2872     cur_test_run.have_churn = HAVE_NO_CHURN;
2873     timeout_s = 10;
2874   }
2875
2876   else if (strstr (argv[0], "_single_peer_seed") != NULL)
2877   {
2878     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Test seeding and requesting on a single peer\n");
2879     cur_test_run.name = "test-rps-single-peer-seed";
2880     cur_test_run.main_test = single_peer_seed_cb;
2881     cur_test_run.have_churn = HAVE_NO_CHURN;
2882   }
2883
2884   else if (strstr (argv[0], "_seed_request") != NULL)
2885   {
2886     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Test seeding and requesting on multiple peers\n");
2887     cur_test_run.name = "test-rps-seed-request";
2888     cur_test_run.main_test = seed_req_cb;
2889     cur_test_run.have_churn = HAVE_NO_CHURN;
2890   }
2891
2892   else if (strstr (argv[0], "_seed") != NULL)
2893   {
2894     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Test seeding\n");
2895     cur_test_run.name = "test-rps-seed";
2896     cur_test_run.main_test = seed_cb;
2897     cur_test_run.eval_cb = no_eval;
2898     cur_test_run.have_churn = HAVE_NO_CHURN;
2899   }
2900
2901   else if (strstr (argv[0], "_req_cancel") != NULL)
2902   {
2903     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Test cancelling a request\n");
2904     cur_test_run.name = "test-rps-req-cancel";
2905     num_peers = 1;
2906     cur_test_run.main_test = req_cancel_cb;
2907     cur_test_run.eval_cb = no_eval;
2908     cur_test_run.have_churn = HAVE_NO_CHURN;
2909     timeout_s = 10;
2910   }
2911
2912   else if (strstr (argv[0], "_churn") != NULL)
2913   {
2914     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Test churn\n");
2915     cur_test_run.name = "test-rps-churn";
2916     num_peers = 5;
2917     cur_test_run.init_peer = default_init_peer;
2918     cur_test_run.main_test = churn_test_cb;
2919     cur_test_run.reply_handle = default_reply_handle;
2920     cur_test_run.eval_cb = default_eval_cb;
2921     cur_test_run.have_churn = HAVE_NO_CHURN;
2922     cur_test_run.have_quick_quit = HAVE_NO_QUICK_QUIT;
2923     timeout_s = 40;
2924   }
2925
2926   else if (strstr (argv[0], "_sub") != NULL)
2927   {
2928     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Test subs\n");
2929     cur_test_run.name = "test-rps-sub";
2930     num_peers = 5;
2931     //cur_test_run.init_peer = &default_init_peer;
2932     cur_test_run.pre_test = &sub_pre;
2933     cur_test_run.main_test = &single_req_cb;
2934     //cur_test_run.reply_handle = default_reply_handle;
2935     cur_test_run.post_test = &sub_post;
2936     //cur_test_run.eval_cb = default_eval_cb;
2937     cur_test_run.have_churn = HAVE_NO_CHURN;
2938     cur_test_run.have_quick_quit = HAVE_QUICK_QUIT;
2939   }
2940
2941   else if (strstr (argv[0], "profiler") != NULL)
2942   {
2943     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "This is the profiler\n");
2944     cur_test_run.name = "test-rps-profiler";
2945     num_peers = 16;
2946     mal_type = 3;
2947     cur_test_run.init_peer = profiler_init_peer;
2948     //cur_test_run.pre_test = mal_pre;
2949     cur_test_run.pre_test = pre_profiler;
2950     cur_test_run.main_test = profiler_cb;
2951     cur_test_run.reply_handle = profiler_reply_handle;
2952     cur_test_run.eval_cb = profiler_eval;
2953     cur_test_run.post_test = post_profiler;
2954     cur_test_run.request_interval = 2;
2955     cur_test_run.num_requests = 5;
2956     //cur_test_run.have_churn = HAVE_CHURN;
2957     cur_test_run.have_churn = HAVE_NO_CHURN;
2958     cur_test_run.have_quick_quit = HAVE_NO_QUICK_QUIT;
2959     cur_test_run.have_collect_statistics = COLLECT_STATISTICS;
2960     cur_test_run.stat_collect_flags = STAT_TYPE_ROUNDS |
2961                                       STAT_TYPE_BLOCKS |
2962                                       STAT_TYPE_BLOCKS_MANY_PUSH |
2963                                       STAT_TYPE_BLOCKS_NO_PUSH |
2964                                       STAT_TYPE_BLOCKS_NO_PULL |
2965                                       STAT_TYPE_BLOCKS_MANY_PUSH_NO_PULL |
2966                                       STAT_TYPE_BLOCKS_NO_PUSH_NO_PULL |
2967                                       STAT_TYPE_ISSUED_PUSH_SEND |
2968                                       STAT_TYPE_ISSUED_PULL_REQ |
2969                                       STAT_TYPE_ISSUED_PULL_REP |
2970                                       STAT_TYPE_SENT_PUSH_SEND |
2971                                       STAT_TYPE_SENT_PULL_REQ |
2972                                       STAT_TYPE_SENT_PULL_REP |
2973                                       STAT_TYPE_RECV_PUSH_SEND |
2974                                       STAT_TYPE_RECV_PULL_REQ |
2975                                       STAT_TYPE_RECV_PULL_REP;
2976     cur_test_run.have_collect_view = COLLECT_VIEW;
2977     timeout_s = 150;
2978
2979     /* 'Clean' directory */
2980     (void) GNUNET_DISK_directory_remove ("/tmp/rps/");
2981     GNUNET_DISK_directory_create ("/tmp/rps/");
2982   }
2983   timeout = GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, timeout_s);
2984
2985   rps_peers = GNUNET_new_array (num_peers, struct RPSPeer);
2986   peer_map = GNUNET_CONTAINER_multipeermap_create (num_peers, GNUNET_NO);
2987   rps_peer_ids = GNUNET_new_array (num_peers, struct GNUNET_PeerIdentity);
2988   if ( (2 == mal_type) ||
2989        (3 == mal_type))
2990     target_peer = &rps_peer_ids[num_peers - 2];
2991   if (profiler_eval == cur_test_run.eval_cb)
2992     eval_peer = &rps_peers[num_peers - 1];    /* FIXME: eval_peer could be a
2993                                                  malicious peer if not careful
2994                                                  with the malicious portion */
2995
2996   ok = 1;
2997   ret_value = GNUNET_TESTBED_test_run (cur_test_run.name,
2998                                        "test_rps.conf",
2999                                        num_peers,
3000                                        0, NULL, NULL,
3001                                        &run, NULL);
3002   GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
3003               "_test_run returned.\n");
3004   if (GNUNET_OK != ret_value)
3005   {
3006     GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
3007                 "Test did not run successfully!\n");
3008   }
3009
3010   ret_value = cur_test_run.eval_cb();
3011
3012   if (NO_COLLECT_VIEW == cur_test_run.have_collect_view)
3013   {
3014     GNUNET_array_grow (rps_peers->cur_view,
3015                        rps_peers->cur_view_count,
3016                        0);
3017   }
3018   GNUNET_free (rps_peers);
3019   GNUNET_free (rps_peer_ids);
3020   GNUNET_CONTAINER_multipeermap_destroy (peer_map);
3021   return ret_value;
3022 }
3023
3024
3025 /* end of test_rps.c */