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