src/cadet/gnunet-cadet-profiler.c

   1 /*
   2      This file is part of GNUnet.
   3      Copyright (C) 2011 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 cadet/gnunet-cadet-profiler.c
  22  *
  23  * @brief Profiler for cadet experiments.
  24  */
  25 #include <stdio.h>
  26 #include "platform.h"
  27 #include "cadet_test_lib.h"
  28 #include "gnunet_cadet_service.h"
  29 #include "gnunet_statistics_service.h"
  30
  31
  32 #define PING 1
  33 #define PONG 2
  34
  35
  36 /**
  37  * Paximum ping period in milliseconds. Real period = rand (0, PING_PERIOD)
  38  */
  39 #define PING_PERIOD 500
  40
  41 /**
  42  * How long until we give up on connecting the peers?
  43  */
  44 #define TIMEOUT GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 120)
  45
  46 /**
  47  * Time to wait for stuff that should be rather fast
  48  */
  49 #define SHORT_TIME GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 300)
  50
  51 /**
  52  * Total number of rounds.
  53  */
  54 #define number_rounds sizeof(rounds) / sizeof(rounds[0])
  55
  56 /**
  57  * Ratio of peers active. First round always is 1.0.
  58  */
  59 static float rounds[] = { 0.8, 0.6, 0.8, 0.5, 0.3, 0.8, 0.0 };
  60
  61 /**
  62  * Message type for pings.
  63  */
  64 struct CadetPingMessage
  65 {
  66   /**
  67    * Header. Type PING/PONG.
  68    */
  69   struct GNUNET_MessageHeader header;
  70
  71   /**
  72    * Message number.
  73    */
  74   uint32_t counter;
  75
  76   /**
  77    * Time the message was sent.
  78    */
  79   struct GNUNET_TIME_AbsoluteNBO timestamp;
  80
  81   /**
  82    * Round number.
  83    */
  84   uint32_t round_number;
  85 };
  86
  87 /**
  88  * Peer description.
  89  */
  90 struct CadetPeer
  91 {
  92   /**
  93    * Testbed Operation (to get peer id, etc).
  94    */
  95   struct GNUNET_TESTBED_Operation *op;
  96
  97   /**
  98    * Peer ID.
  99    */
 100   struct GNUNET_PeerIdentity id;
 101
 102   /**
 103    * Cadet handle for the root peer
 104    */
 105   struct GNUNET_CADET_Handle *cadet;
 106
 107   /**
 108    * Channel handle for the root peer
 109    */
 110   struct GNUNET_CADET_Channel *ch;
 111
 112   /**
 113    * Channel handle for the dest peer
 114    */
 115   struct GNUNET_CADET_Channel *incoming_ch;
 116
 117   /**
 118    * Channel handle for a warmup channel.
 119    */
 120   struct GNUNET_CADET_Channel *warmup_ch;
 121
 122   /**
 123    * Number of payload packes sent
 124    */
 125   int data_sent;
 126
 127   /**
 128    * Number of payload packets received
 129    */
 130   int data_received;
 131
 132   /**
 133    * Is peer up?
 134    */
 135   int up;
 136
 137   /**
 138    * Destinaton to ping.
 139    */
 140   struct CadetPeer *dest;
 141
 142   /**
 143    * Incoming channel for pings.
 144    */
 145   struct CadetPeer *incoming;
 146
 147   /**
 148    * Task to do the next ping.
 149    */
 150   struct GNUNET_SCHEDULER_Task *ping_task;
 151
 152   /**
 153    * NTR operation for the next ping.
 154    */
 155   struct GNUNET_CADET_TransmitHandle *ping_ntr;
 156
 157   float mean[number_rounds];
 158   float var[number_rounds];
 159   unsigned int pongs[number_rounds];
 160   unsigned int pings[number_rounds];
 161 };
 162
 163 /**
 164  * Duration of each round.
 165  */
 166 static struct GNUNET_TIME_Relative round_time;
 167
 168 /**
 169  * GNUNET_PeerIdentity -> CadetPeer
 170  */
 171 static struct GNUNET_CONTAINER_MultiPeerMap *ids;
 172
 173 /**
 174  * Testbed peer handles.
 175  */
 176 static struct GNUNET_TESTBED_Peer **testbed_handles;
 177
 178 /**
 179  * Testbed Operation (to get stats).
 180  */
 181 static struct GNUNET_TESTBED_Operation *stats_op;
 182
 183 /**
 184  * Operation to get peer ids.
 185  */
 186 static struct CadetPeer *peers;
 187
 188 /**
 189  * Peer ids counter.
 190  */
 191 static unsigned int p_ids;
 192
 193 /**
 194  * Total number of peers.
 195  */
 196 static unsigned long long peers_total;
 197
 198 /**
 199  * Number of currently running peers.
 200  */
 201 static unsigned long long peers_running;
 202
 203 /**
 204  * Number of peers doing pings.
 205  */
 206 static unsigned long long peers_pinging;
 207
 208 /**
 209  * Test context (to shut down).
 210  */
 211 static struct GNUNET_CADET_TEST_Context *test_ctx;
 212
 213 /**
 214  * Task called to disconnect peers, before shutdown.
 215  */
 216 static struct GNUNET_SCHEDULER_Task *disconnect_task;
 217
 218 /**
 219  * Task to perform tests
 220  */
 221 static struct GNUNET_SCHEDULER_Task *test_task;
 222
 223 /**
 224  * Round number.
 225  */
 226 static unsigned int current_round;
 227
 228 /**
 229  * Do preconnect? (Each peer creates a tunnel to one other peer).
 230  */
 231 static int do_warmup;
 232
 233 /**
 234  * Warmup progress.
 235  */
 236 static unsigned int peers_warmup;
 237
 238 /**
 239  * Flag to notify callbacks not to generate any new traffic anymore.
 240  */
 241 static int test_finished;
 242
 243 /**
 244  * Task running each round of the benchmark.
 245  */
 246 static struct GNUNET_SCHEDULER_Task *round_task;
 247
 248
 249 /**
 250  * START THE TEST ITSELF, AS WE ARE CONNECTED TO THE CADET SERVICES.
 251  *
 252  * Testcase continues when the root receives confirmation of connected peers,
 253  * on callback funtion ch.
 254  *
 255  * @param cls Closure (unsued).
 256  */
 257 static void
 258 start_test (void *cls);
 259
 260
 261 /**
 262  * Calculate a random delay.
 263  *
 264  * @param max Exclusive maximum, in ms.
 265  *
 266  * @return A time between 0 a max-1 ms.
 267  */
 268 static struct GNUNET_TIME_Relative
 269 delay_ms_rnd (unsigned int max)
 270 {
 271   unsigned int rnd;
 272
 273   rnd = GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK, max);
 274   return GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MILLISECONDS, rnd);
 275 }
 276
 277
 278 /**
 279  * Get the index of a peer in the peers array.
 280  *
 281  * @param peer Peer whose index to get.
 282  *
 283  * @return Index of peer in peers.
 284  */
 285 static unsigned int
 286 get_index (struct CadetPeer *peer)
 287 {
 288   return peer - peers;
 289 }
 290
 291
 292 /**
 293  * Show the results of the test (banwidth acheived) and log them to GAUGER
 294  */
 295 static void
 296 show_end_data (void)
 297 {
 298   struct CadetPeer *peer;
 299   unsigned int i;
 300   unsigned int j;
 301
 302   for (i = 0; i < number_rounds; i++)
 303   {
 304     for (j = 0; j < peers_pinging; j++)
 305     {
 306       peer = &peers[j];
 307       fprintf (stdout,
 308                "ROUND %3u PEER %3u: %10.2f / %10.2f, PINGS: %3u, PONGS: %3u\n",
 309                i, j, peer->mean[i], sqrt (peer->var[i] / (peer->pongs[i] - 1)),
 310                peer->pings[i], peer->pongs[i]);
 311     }
 312   }
 313 }
 314
 315
 316 /**
 317  * Disconnect from cadet services af all peers, call shutdown.
 318  *
 319  * @param cls Closure (unused).
 320  */
 321 static void
 322 disconnect_cadet_peers (void *cls)
 323 {
 324   long line = (long) cls;
 325   unsigned int i;
 326
 327   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 328               "disconnecting cadet service, called from line %ld\n",
 329               line);
 330   disconnect_task = NULL;
 331   for (i = 0; i < peers_total; i++)
 332   {
 333     if (NULL != peers[i].op)
 334       GNUNET_TESTBED_operation_done (peers[i].op);
 335
 336     if (peers[i].up != GNUNET_YES)
 337       continue;
 338
 339     if (NULL != peers[i].ch)
 340     {
 341       GNUNET_log (GNUNET_ERROR_TYPE_INFO,
 342                   "%u: channel %p\n", i, peers[i].ch);
 343       GNUNET_CADET_channel_destroy (peers[i].ch);
 344     }
 345     if (NULL != peers[i].warmup_ch)
 346     {
 347       GNUNET_log (GNUNET_ERROR_TYPE_INFO,
 348                   "%u: warmup channel %p\n",
 349                   i, peers[i].warmup_ch);
 350       GNUNET_CADET_channel_destroy (peers[i].warmup_ch);
 351     }
 352     if (NULL != peers[i].incoming_ch)
 353     {
 354       GNUNET_log (GNUNET_ERROR_TYPE_INFO,
 355                   "%u: incoming channel %p\n",
 356                   i, peers[i].incoming_ch);
 357       GNUNET_CADET_channel_destroy (peers[i].incoming_ch);
 358     }
 359   }
 360   GNUNET_CADET_TEST_cleanup (test_ctx);
 361   GNUNET_SCHEDULER_shutdown ();
 362 }
 363
 364
 365 /**
 366  * Shut down peergroup, clean up.
 367  *
 368  * @param cls Closure (unused).
 369  */
 370 static void
 371 shutdown_task (void *cls)
 372 {
 373   GNUNET_log (GNUNET_ERROR_TYPE_INFO,
 374               "Ending test.\n");
 375   if (NULL != disconnect_task)
 376   {
 377     GNUNET_SCHEDULER_cancel (disconnect_task);
 378     disconnect_task = GNUNET_SCHEDULER_add_now (&disconnect_cadet_peers,
 379                                                 (void *) __LINE__);
 380   }
 381   if (NULL != round_task)
 382   {
 383     GNUNET_SCHEDULER_cancel (round_task);
 384     round_task = NULL;
 385   }
 386   if (NULL != test_task)
 387   {
 388     GNUNET_SCHEDULER_cancel (test_task);
 389     test_task = NULL;
 390   }
 391 }
 392
 393
 394 /**
 395  * Finish test normally: schedule disconnect and shutdown
 396  *
 397  * @param line Line in the code the abort is requested from (__LINE__).
 398  */
 399 static void
 400 abort_test (long line)
 401 {
 402   if (disconnect_task != NULL)
 403   {
 404     GNUNET_SCHEDULER_cancel (disconnect_task);
 405     disconnect_task = GNUNET_SCHEDULER_add_now (&disconnect_cadet_peers,
 406                                                 (void *) line);
 407   }
 408 }
 409
 410
 411 /**
 412  * Stats callback. Finish the stats testbed operation and when all stats have
 413  * been iterated, shutdown the test.
 414  *
 415  * @param cls closure
 416  * @param op the operation that has been finished
 417  * @param emsg error message in case the operation has failed; will be NULL if
 418  *          operation has executed successfully.
 419  */
 420 static void
 421 stats_cont (void *cls, struct GNUNET_TESTBED_Operation *op, const char *emsg)
 422 {
 423   GNUNET_log (GNUNET_ERROR_TYPE_INFO, "... collecting statistics done.\n");
 424   GNUNET_TESTBED_operation_done (stats_op);
 425
 426   if (NULL != disconnect_task)
 427     GNUNET_SCHEDULER_cancel (disconnect_task);
 428   disconnect_task = GNUNET_SCHEDULER_add_now (&disconnect_cadet_peers,
 429                                               (void *) __LINE__);
 430 }
 431
 432
 433 /**
 434  * Process statistic values.
 435  *
 436  * @param cls closure
 437  * @param peer the peer the statistic belong to
 438  * @param subsystem name of subsystem that created the statistic
 439  * @param name the name of the datum
 440  * @param value the current value
 441  * @param is_persistent #GNUNET_YES if the value is persistent, #GNUNET_NO if not
 442  * @return #GNUNET_OK to continue, #GNUNET_SYSERR to abort iteration
 443  */
 444 static int
 445 stats_iterator (void *cls,
 446                 const struct GNUNET_TESTBED_Peer *peer,
 447                 const char *subsystem,
 448                 const char *name,
 449                 uint64_t value,
 450                 int is_persistent)
 451 {
 452   uint32_t i;
 453
 454   i = GNUNET_TESTBED_get_index (peer);
 455   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 456               " STATS %u - %s [%s]: %llu\n",
 457               i, subsystem, name,
 458               (unsigned long long) value);
 459
 460   return GNUNET_OK;
 461 }
 462
 463
 464 /**
 465  * Task check that keepalives were sent and received.
 466  *
 467  * @param cls Closure (NULL).
 468  */
 469 static void
 470 collect_stats (void *cls)
 471 {
 472   GNUNET_log (GNUNET_ERROR_TYPE_INFO,
 473               "Start collecting statistics...\n");
 474   stats_op = GNUNET_TESTBED_get_statistics (peers_total,
 475                                             testbed_handles,
 476                                             NULL, NULL,
 477                                             &stats_iterator,
 478                                             &stats_cont, NULL);
 479 }
 480
 481
 482 /**
 483  * @brief Finish profiler normally. Signal finish and start collecting stats.
 484  *
 485  * @param cls Closure (unused).
 486  */
 487 static void
 488 finish_profiler (void *cls)
 489 {
 490   test_finished = GNUNET_YES;
 491   show_end_data ();
 492   GNUNET_SCHEDULER_add_now (&collect_stats, NULL);
 493 }
 494
 495
 496 /**
 497  * Set the total number of running peers.
 498  *
 499  * @param target Desired number of running peers.
 500  */
 501 static void
 502 adjust_running_peers (unsigned int target)
 503 {
 504   struct GNUNET_TESTBED_Operation *op;
 505   unsigned int delta;
 506   unsigned int run;
 507   unsigned int i;
 508   unsigned int r;
 509
 510   GNUNET_assert (target <= peers_total);
 511
 512   GNUNET_log (GNUNET_ERROR_TYPE_INFO, "adjust peers to %u\n", target);
 513   if (target > peers_running)
 514   {
 515     delta = target - peers_running;
 516     run = GNUNET_YES;
 517   }
 518   else
 519   {
 520     delta = peers_running - target;
 521     run = GNUNET_NO;
 522   }
 523
 524   for (i = 0; i < delta; i++)
 525   {
 526     do
 527     {
 528       r = GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
 529                                     peers_total - peers_pinging);
 530       r += peers_pinging;
 531     }
 532     while (peers[r].up == run || NULL != peers[r].incoming);
 533     GNUNET_log (GNUNET_ERROR_TYPE_INFO, "St%s peer %u: %s\n",
 534                 run ? "arting" : "opping", r, GNUNET_i2s (&peers[r].id));
 535
 536     if (NULL != peers[r].ping_task)
 537     {
 538       GNUNET_SCHEDULER_cancel (peers[r].ping_task);
 539       peers[r].ping_task = NULL;
 540     }
 541     if (NULL != peers[r].ping_ntr)
 542     {
 543       GNUNET_CADET_notify_transmit_ready_cancel (peers[r].ping_ntr);
 544       peers[r].ping_ntr = NULL;
 545     }
 546     peers[r].up = run;
 547
 548     if (NULL != peers[r].ch)
 549       GNUNET_CADET_channel_destroy (peers[r].ch);
 550     peers[r].ch = NULL;
 551     if (NULL != peers[r].dest)
 552     {
 553       if (NULL != peers[r].dest->incoming_ch)
 554         GNUNET_CADET_channel_destroy (peers[r].dest->incoming_ch);
 555       peers[r].dest->incoming_ch = NULL;
 556     }
 557
 558     op = GNUNET_TESTBED_peer_manage_service (&peers[r], testbed_handles[r],
 559                                              "cadet", NULL, NULL, run);
 560     GNUNET_break (NULL != op);
 561     peers_running += run ? 1 : -1;
 562     GNUNET_assert (peers_running > 0);
 563   }
 564 }
 565
 566
 567 /**
 568  * @brief Move to next round.
 569  *
 570  * @param cls Closure (round #).
 571  */
 572 static void
 573 next_rnd (void *cls)
 574 {
 575   GNUNET_log (GNUNET_ERROR_TYPE_INFO,
 576               "ROUND %u\n",
 577               current_round);
 578   if (0.0 == rounds[current_round])
 579   {
 580     GNUNET_log (GNUNET_ERROR_TYPE_INFO, "Finishing\n");
 581     GNUNET_SCHEDULER_add_now (&finish_profiler, NULL);
 582     return;
 583   }
 584   adjust_running_peers (rounds[current_round] * peers_total);
 585   current_round++;
 586
 587   round_task = GNUNET_SCHEDULER_add_delayed (round_time,
 588                                              &next_rnd,
 589                                              NULL);
 590 }
 591
 592
 593 /**
 594  * Transmit ping callback.
 595  *
 596  * @param cls Closure (peer for PING, NULL for PONG).
 597  * @param size Size of the tranmist buffer.
 598  * @param buf Pointer to the beginning of the buffer.
 599  *
 600  * @return Number of bytes written to buf.
 601  */
 602 static size_t
 603 tmt_rdy_ping (void *cls, size_t size, void *buf);
 604
 605
 606 /**
 607  * Transmit pong callback.
 608  *
 609  * @param cls Closure (copy of PING message, to be freed).
 610  * @param size Size of the buffer we have.
 611  * @param buf Buffer to copy data to.
 612  */
 613 static size_t
 614 tmt_rdy_pong (void *cls, size_t size, void *buf)
 615 {
 616   struct CadetPingMessage *ping = cls;
 617   struct CadetPingMessage *pong;
 618
 619   if ((0 == size) || (NULL == buf))
 620   {
 621     GNUNET_free (ping);
 622     return 0;
 623   }
 624   pong = (struct CadetPingMessage *) buf;
 625   GNUNET_memcpy (pong, ping, sizeof(*ping));
 626   pong->header.type = htons (PONG);
 627
 628   GNUNET_free (ping);
 629   return sizeof(*ping);
 630 }
 631
 632
 633 /**
 634  * @brief Send a ping to destination
 635  *
 636  * @param cls Closure (peer).
 637  */
 638 static void
 639 ping (void *cls)
 640 {
 641   struct CadetPeer *peer = cls;
 642
 643   peer->ping_task = NULL;
 644   if (GNUNET_YES == test_finished)
 645     return;
 646   GNUNET_log (GNUNET_ERROR_TYPE_INFO,
 647               "%u -> %u (%u)\n",
 648               get_index (peer),
 649               get_index (peer->dest),
 650               peer->data_sent);
 651   peer->ping_ntr = GNUNET_CADET_notify_transmit_ready (peer->ch, GNUNET_NO,
 652                                                        GNUNET_TIME_UNIT_FOREVER_REL,
 653                                                        sizeof(struct
 654                                                               CadetPingMessage),
 655                                                        &tmt_rdy_ping, peer);
 656 }
 657
 658
 659 /**
 660  * @brief Reply with a pong to origin.
 661  *
 662  * @param cls Closure (peer).
 663  * @param tc Task context.
 664  */
 665 static void
 666 pong (struct GNUNET_CADET_Channel *channel,
 667       const struct CadetPingMessage *ping)
 668 {
 669   struct CadetPingMessage *copy;
 670
 671   copy = GNUNET_new (struct CadetPingMessage);
 672   *copy = *ping;
 673   GNUNET_CADET_notify_transmit_ready (channel, GNUNET_NO,
 674                                       GNUNET_TIME_UNIT_FOREVER_REL,
 675                                       sizeof(struct CadetPingMessage),
 676                                       &tmt_rdy_pong, copy);
 677 }
 678
 679
 680 /**
 681  * Transmit ping callback
 682  *
 683  * @param cls Closure (peer).
 684  * @param size Size of the buffer we have.
 685  * @param buf Buffer to copy data to.
 686  */
 687 static size_t
 688 tmt_rdy_ping (void *cls, size_t size, void *buf)
 689 {
 690   struct CadetPeer *peer = cls;
 691   struct CadetPingMessage *msg = buf;
 692
 693   peer->ping_ntr = NULL;
 694   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 695               "tmt_rdy called, filling buffer\n");
 696   if ((size < sizeof(struct CadetPingMessage)) || (NULL == buf))
 697   {
 698     GNUNET_break (GNUNET_YES == test_finished);
 699     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 700                 "size %u, buf %p, data_sent %u, data_received %u\n",
 701                 (unsigned int) size,
 702                 buf,
 703                 peer->data_sent,
 704                 peer->data_received);
 705
 706     return 0;
 707   }
 708   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 709               "Sending: msg %d\n",
 710               peer->data_sent);
 711   msg->header.size = htons (size);
 712   msg->header.type = htons (PING);
 713   msg->counter = htonl (peer->data_sent++);
 714   msg->round_number = htonl (current_round);
 715   msg->timestamp = GNUNET_TIME_absolute_hton (GNUNET_TIME_absolute_get ());
 716   peer->pings[current_round]++;
 717   peer->ping_task = GNUNET_SCHEDULER_add_delayed (delay_ms_rnd (PING_PERIOD),
 718                                                   &ping, peer);
 719
 720   return sizeof(struct CadetPingMessage);
 721 }
 722
 723
 724 /**
 725  * Function is called whenever a PING message is received.
 726  *
 727  * @param cls closure (peer #, set from GNUNET_CADET_connect)
 728  * @param channel connection to the other end
 729  * @param channel_ctx place to store local state associated with the channel
 730  * @param message the actual message
 731  * @return GNUNET_OK to keep the connection open,
 732  *         GNUNET_SYSERR to close it (signal serious error)
 733  */
 734 int
 735 ping_handler (void *cls, struct GNUNET_CADET_Channel *channel,
 736               void **channel_ctx,
 737               const struct GNUNET_MessageHeader *message)
 738 {
 739   long n = (long) cls;
 740
 741   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 742               "%u got PING\n",
 743               (unsigned int) n);
 744   GNUNET_CADET_receive_done (channel);
 745   if (GNUNET_NO == test_finished)
 746     pong (channel, (struct CadetPingMessage *) message);
 747
 748   return GNUNET_OK;
 749 }
 750
 751
 752 /**
 753  * Function is called whenever a PONG message is received.
 754  *
 755  * @param cls closure (peer #, set from GNUNET_CADET_connect)
 756  * @param channel connection to the other end
 757  * @param channel_ctx place to store local state associated with the channel
 758  * @param message the actual message
 759  * @return GNUNET_OK to keep the connection open,
 760  *         GNUNET_SYSERR to close it (signal serious error)
 761  */
 762 int
 763 pong_handler (void *cls, struct GNUNET_CADET_Channel *channel,
 764               void **channel_ctx,
 765               const struct GNUNET_MessageHeader *message)
 766 {
 767   long n = (long) cls;
 768   struct CadetPeer *peer;
 769   struct CadetPingMessage *msg;
 770   struct GNUNET_TIME_Absolute send_time;
 771   struct GNUNET_TIME_Relative latency;
 772   unsigned int r /* Ping round */;
 773   float delta;
 774
 775   GNUNET_CADET_receive_done (channel);
 776   peer = &peers[n];
 777
 778   msg = (struct CadetPingMessage *) message;
 779
 780   send_time = GNUNET_TIME_absolute_ntoh (msg->timestamp);
 781   latency = GNUNET_TIME_absolute_get_duration (send_time);
 782   r = ntohl (msg->round_number);
 783   GNUNET_log (GNUNET_ERROR_TYPE_INFO, "%u <- %u (%u) latency: %s\n",
 784               get_index (peer),
 785               get_index (peer->dest),
 786               (uint32_t) ntohl (msg->counter),
 787               GNUNET_STRINGS_relative_time_to_string (latency, GNUNET_NO));
 788
 789   /* Online variance calculation */
 790   peer->pongs[r]++;
 791   delta = latency.rel_value_us - peer->mean[r];
 792   peer->mean[r] = peer->mean[r] + delta / peer->pongs[r];
 793   peer->var[r] += delta * (latency.rel_value_us - peer->mean[r]);
 794
 795   return GNUNET_OK;
 796 }
 797
 798
 799 /**
 800  * Handlers, for diverse services
 801  */
 802 static struct GNUNET_CADET_MessageHandler handlers[] = {
 803   { &ping_handler, PING, sizeof(struct CadetPingMessage) },
 804   { &pong_handler, PONG, sizeof(struct CadetPingMessage) },
 805   { NULL, 0, 0 }
 806 };
 807
 808
 809 /**
 810  * Method called whenever another peer has added us to a channel
 811  * the other peer initiated.
 812  *
 813  * @param cls Closure.
 814  * @param channel New handle to the channel.
 815  * @param initiator Peer that started the channel.
 816  * @param port Port this channel is connected to.
 817  * @param options channel option flags
 818  * @return Initial channel context for the channel
 819  *         (can be NULL -- that's not an error).
 820  */
 821 static void *
 822 incoming_channel (void *cls, struct GNUNET_CADET_Channel *channel,
 823                   const struct GNUNET_PeerIdentity *initiator,
 824                   const struct GNUNET_HashCode *port,
 825                   enum GNUNET_CADET_ChannelOption options)
 826 {
 827   long n = (long) cls;
 828   struct CadetPeer *peer;
 829
 830   peer = GNUNET_CONTAINER_multipeermap_get (ids, initiator);
 831   GNUNET_assert (NULL != peer);
 832   if (NULL == peers[n].incoming)
 833   {
 834     GNUNET_log (GNUNET_ERROR_TYPE_INFO,
 835                 "WARMUP %3u: %u <= %u\n",
 836                 peers_warmup,
 837                 (unsigned int) n,
 838                 get_index (peer));
 839     peers_warmup++;
 840     if (peers_warmup < peers_total)
 841       return NULL;
 842     if (NULL != test_task)
 843     {
 844       GNUNET_SCHEDULER_cancel (test_task);
 845       test_task = GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_SECONDS,
 846                                                 &start_test, NULL);
 847     }
 848     return NULL;
 849   }
 850   GNUNET_assert (peer == peers[n].incoming);
 851   GNUNET_assert (peer->dest == &peers[n]);
 852   GNUNET_log (GNUNET_ERROR_TYPE_INFO,
 853               "%u <= %u %p\n",
 854               (unsigned int) n,
 855               get_index (peer),
 856               channel);
 857   peers[n].incoming_ch = channel;
 858
 859   return NULL;
 860 }
 861
 862
 863 /**
 864  * Function called whenever an inbound channel is destroyed.  Should clean up
 865  * any associated state.
 866  *
 867  * @param cls closure (set from GNUNET_CADET_connect)
 868  * @param channel connection to the other end (henceforth invalid)
 869  * @param channel_ctx place where local state associated
 870  *                   with the channel is stored
 871  */
 872 static void
 873 channel_cleaner (void *cls,
 874                  const struct GNUNET_CADET_Channel *channel,
 875                  void *channel_ctx)
 876 {
 877   long n = (long) cls;
 878   struct CadetPeer *peer = &peers[n];
 879
 880   GNUNET_log (GNUNET_ERROR_TYPE_INFO,
 881               "Channel %p disconnected at peer %ld\n", channel, n);
 882   if (peer->ch == channel)
 883     peer->ch = NULL;
 884 }
 885
 886
 887 /**
 888  * Select a random peer that has no incoming channel
 889  *
 890  * @param peer ID of the peer connecting. NULL if irrelevant (warmup).
 891  *
 892  * @return Random peer not yet connected to.
 893  */
 894 static struct CadetPeer *
 895 select_random_peer (struct CadetPeer *peer)
 896 {
 897   unsigned int r;
 898
 899   do
 900   {
 901     r = GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK, peers_total);
 902   }
 903   while (NULL != peers[r].incoming);
 904   peers[r].incoming = peer;
 905
 906   return &peers[r];
 907 }
 908
 909
 910 /**
 911  * START THE TEST ITSELF, AS WE ARE CONNECTED TO THE CADET SERVICES.
 912  *
 913  * Testcase continues when the root receives confirmation of connected peers,
 914  * on callback funtion ch.
 915  *
 916  * @param cls Closure (unsued).
 917  */
 918 static void
 919 start_test (void *cls)
 920 {
 921   unsigned long i;
 922
 923   test_task = NULL;
 924   GNUNET_log (GNUNET_ERROR_TYPE_INFO, "Start profiler\n");
 925
 926
 927   for (i = 0; i < peers_pinging; i++)
 928   {
 929     peers[i].dest = select_random_peer (&peers[i]);
 930     peers[i].ch = GNUNET_CADET_channel_create (peers[i].cadet, NULL,
 931                                                &peers[i].dest->id,
 932                                                GC_u2h (1));
 933     if (NULL == peers[i].ch)
 934     {
 935       GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "Channel %lu failed\n", i);
 936       GNUNET_CADET_TEST_cleanup (test_ctx);
 937       return;
 938     }
 939     GNUNET_log (GNUNET_ERROR_TYPE_INFO,
 940                 "%lu => %u %p\n",
 941                 i,
 942                 get_index (peers[i].dest),
 943                 peers[i].ch);
 944     peers[i].ping_task = GNUNET_SCHEDULER_add_delayed (delay_ms_rnd (2000),
 945                                                        &ping, &peers[i]);
 946   }
 947   peers_running = peers_total;
 948   if (NULL != disconnect_task)
 949     GNUNET_SCHEDULER_cancel (disconnect_task);
 950   disconnect_task =
 951     GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_relative_multiply (round_time,
 952                                                                  number_rounds
 953                                                                  + 1),
 954                                   &disconnect_cadet_peers,
 955                                   (void *) __LINE__);
 956   round_task = GNUNET_SCHEDULER_add_delayed (round_time,
 957                                              &next_rnd,
 958                                              NULL);
 959 }
 960
 961
 962 /**
 963  * Do warmup: create some channels to spread information about the topology.
 964  */
 965 static void
 966 warmup (void)
 967 {
 968   struct CadetPeer *peer;
 969   unsigned int i;
 970
 971   for (i = 0; i < peers_total; i++)
 972   {
 973     peer = select_random_peer (NULL);
 974     GNUNET_log (GNUNET_ERROR_TYPE_INFO, "WARMUP %u => %u\n",
 975                 i, get_index (peer));
 976     peers[i].warmup_ch =
 977       GNUNET_CADET_channel_create (peers[i].cadet, NULL, &peer->id,
 978                                    GC_u2h (1));
 979     if (NULL == peers[i].warmup_ch)
 980     {
 981       GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "Warmup %u failed\n", i);
 982       GNUNET_CADET_TEST_cleanup (test_ctx);
 983       return;
 984     }
 985   }
 986 }
 987
 988
 989 /**
 990  * Callback to be called when the requested peer information is available
 991  *
 992  * @param cls the closure from GNUNET_TESTBED_peer_get_information()
 993  * @param op the operation this callback corresponds to
 994  * @param pinfo the result; will be NULL if the operation has failed
 995  * @param emsg error message if the operation has failed;
 996  *             NULL if the operation is successfull
 997  */
 998 static void
 999 peer_id_cb (void *cls,
1000             struct GNUNET_TESTBED_Operation *op,
1001             const struct GNUNET_TESTBED_PeerInformation *pinfo,
1002             const char *emsg)
1003 {
1004   long n = (long) cls;
1005
1006   if ((NULL == pinfo) || (NULL != emsg))
1007   {
1008     GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "pi_cb: %s\n", emsg);
1009     abort_test (__LINE__);
1010     return;
1011   }
1012   peers[n].id = *(pinfo->result.id);
1013   GNUNET_log (GNUNET_ERROR_TYPE_INFO,
1014               "%ld  id: %s\n",
1015               n,
1016               GNUNET_i2s (&peers[n].id));
1017   GNUNET_break (GNUNET_OK ==
1018                 GNUNET_CONTAINER_multipeermap_put (ids, &peers[n].id, &peers[n],
1019                                                    GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_FAST));
1020
1021   GNUNET_TESTBED_operation_done (peers[n].op);
1022   peers[n].op = NULL;
1023
1024   p_ids++;
1025   if (p_ids < peers_total)
1026     return;
1027   GNUNET_log (GNUNET_ERROR_TYPE_INFO, "Got all IDs, starting profiler\n");
1028   if (do_warmup)
1029   {
1030     struct GNUNET_TIME_Relative delay;
1031
1032     warmup ();
1033     delay = GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MILLISECONDS,
1034                                            100 * peers_total);
1035     test_task = GNUNET_SCHEDULER_add_delayed (delay, &start_test, NULL);
1036     return;   /* start_test from incoming_channel */
1037   }
1038   GNUNET_log (GNUNET_ERROR_TYPE_INFO, "Starting in a second...\n");
1039   test_task = GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_SECONDS,
1040                                             &start_test, NULL);
1041 }
1042
1043
1044 /**
1045  * test main: start test when all peers are connected
1046  *
1047  * @param cls Closure.
1048  * @param ctx Argument to give to GNUNET_CADET_TEST_cleanup on test end.
1049  * @param num_peers Number of peers that are running.
1050  * @param testbed_peers Array of peers.
1051  * @param cadetes Handle to each of the CADETs of the peers.
1052  */
1053 static void
1054 tmain (void *cls,
1055        struct GNUNET_CADET_TEST_Context *ctx,
1056        unsigned int num_peers,
1057        struct GNUNET_TESTBED_Peer **testbed_peers,
1058        struct GNUNET_CADET_Handle **cadetes)
1059 {
1060   unsigned long i;
1061
1062   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1063               "test main\n");
1064   test_ctx = ctx;
1065   GNUNET_assert (peers_total == num_peers);
1066   peers_running = num_peers;
1067   testbed_handles = testbed_peers;
1068   disconnect_task = GNUNET_SCHEDULER_add_delayed (SHORT_TIME,
1069                                                   &disconnect_cadet_peers,
1070                                                   (void *) __LINE__);
1071   GNUNET_SCHEDULER_add_shutdown (&shutdown_task, NULL);
1072   for (i = 0; i < peers_total; i++)
1073   {
1074     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1075                 "requesting id %ld\n",
1076                 i);
1077     peers[i].up = GNUNET_YES;
1078     peers[i].cadet = cadetes[i];
1079     peers[i].op =
1080       GNUNET_TESTBED_peer_get_information (testbed_handles[i],
1081                                            GNUNET_TESTBED_PIT_IDENTITY,
1082                                            &peer_id_cb, (void *) i);
1083   }
1084   GNUNET_log (GNUNET_ERROR_TYPE_INFO, "requested peer ids\n");
1085   /* Continues from pi_cb -> do_test */
1086 }
1087
1088
1089 /**
1090  * Main: start profiler.
1091  */
1092 int
1093 main (int argc, char *argv[])
1094 {
1095   static const struct GNUNET_HashCode *ports[2];
1096   const char *config_file;
1097
1098   config_file = ".profiler.conf";
1099
1100   if (4 > argc)
1101   {
1102     fprintf (stderr,
1103              "usage: %s ROUND_TIME PEERS PINGS [DO_WARMUP]\n",
1104              argv[0]);
1105     fprintf (stderr,
1106              "example: %s 30s 16 1 Y\n",
1107              argv[0]);
1108     return 1;
1109   }
1110
1111   if (GNUNET_OK !=
1112       GNUNET_STRINGS_fancy_time_to_relative (argv[1],
1113                                              &round_time))
1114   {
1115     fprintf (stderr,
1116              "%s is not a valid time\n",
1117              argv[1]);
1118     return 1;
1119   }
1120
1121   peers_total = atoll (argv[2]);
1122   if (2 > peers_total)
1123   {
1124     fprintf (stderr,
1125              "%s peers is not valid (> 2)\n",
1126              argv[1]);
1127     return 1;
1128   }
1129   peers = GNUNET_new_array (peers_total,
1130                             struct CadetPeer);
1131   peers_pinging = atoll (argv[3]);
1132
1133   if (peers_total < 2 * peers_pinging)
1134   {
1135     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
1136                 "not enough peers, total should be > 2 * peers_pinging\n");
1137     return 1;
1138   }
1139
1140   do_warmup = (5 > argc || argv[4][0] != 'N');
1141
1142   ids = GNUNET_CONTAINER_multipeermap_create (2 * peers_total,
1143                                               GNUNET_YES);
1144   GNUNET_assert (NULL != ids);
1145   p_ids = 0;
1146   test_finished = GNUNET_NO;
1147   ports[0] = GC_u2h (1);
1148   ports[1] = 0;
1149   GNUNET_CADET_TEST_run ("cadet-profiler", config_file, peers_total,
1150                          &tmain, NULL, /* tmain cls */
1151                          &incoming_channel, &channel_cleaner,
1152                          handlers, ports);
1153   GNUNET_free (peers);
1154
1155   return 0;
1156 }
1157
1158
1159 /* end of gnunet-cadet-profiler.c */