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