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