src/cadet/gnunet-cadet-profiler.c

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