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