src/mesh/gnunet-mesh-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 mesh/gnunet-mesh-profiler.c
  22  *
  23  * @brief Profiler for mesh experiments.
  24  */
  25 #include <stdio.h>
  26 #include "platform.h"
  27 #include "mesh_test_lib.h"
  28 #include "gnunet_mesh_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, 60)
  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 MeshPingMessage
  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 MeshPeer
  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    * Mesh handle for the root peer
 104    */
 105   struct GNUNET_MESH_Handle *mesh;
 106
 107   /**
 108    * Channel handle for the root peer
 109    */
 110   struct GNUNET_MESH_Channel *ch;
 111
 112   /**
 113    * Channel handle for the dest peer
 114    */
 115   struct GNUNET_MESH_Channel *incoming_ch;
 116
 117   /**
 118    * Channel handle for a warmup channel.
 119    */
 120   struct GNUNET_MESH_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 MeshPeer *dest;
 141
 142   /**
 143    * Incoming channel for pings.
 144    */
 145   struct MeshPeer *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 -> MeshPeer
 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 MeshPeer *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_MESH_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 MESH 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 MeshPeer *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 MeshPeer *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 mesh services af all peers, call shutdown.
 329  *
 330  * @param cls Closure (unused).
 331  * @param tc Task Context.
 332  */
 333 static void
 334 disconnect_mesh_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 mesh 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_MESH_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_MESH_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_MESH_channel_destroy (peers[i].incoming_ch);
 366     }
 367   }
 368   GNUNET_MESH_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_mesh_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_mesh_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_MESH_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_MESH_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                                              "mesh", 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 MeshPingMessage *ping = cls;
 591   struct MeshPingMessage *pong;
 592
 593   if (0 == size || NULL == buf)
 594   {
 595     GNUNET_free (ping);
 596     return 0;
 597   }
 598   pong = (struct MeshPingMessage *) 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 MeshPeer *peer = (struct MeshPeer *) 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_MESH_notify_transmit_ready (peer->ch, GNUNET_NO,
 628                                      GNUNET_TIME_UNIT_FOREVER_REL,
 629                                      sizeof (struct MeshPingMessage),
 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_MESH_Channel *channel, const struct MeshPingMessage *ping)
 641 {
 642   struct MeshPingMessage *copy;
 643
 644   copy = GNUNET_new (struct MeshPingMessage);
 645   memcpy (copy, ping, sizeof (*ping));
 646   GNUNET_MESH_notify_transmit_ready (channel, GNUNET_NO,
 647                                      GNUNET_TIME_UNIT_FOREVER_REL,
 648                                      sizeof (struct MeshPingMessage),
 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 MeshPeer *peer = (struct MeshPeer *) cls;
 664   struct MeshPingMessage *msg = buf;
 665
 666   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "tmt_rdy called, filling buffer\n");
 667   if (size < sizeof (struct MeshPingMessage) || NULL == buf)
 668   {
 669     GNUNET_break (0);
 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 MeshPingMessage);
 687 }
 688
 689
 690 /**
 691  * Function is called whenever a PING message is received.
 692  *
 693  * @param cls closure (peer #, set from GNUNET_MESH_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_MESH_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_MESH_receive_done (channel);
 709   if (GNUNET_NO == test_finished)
 710     pong (channel, (struct MeshPingMessage *) 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_MESH_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_MESH_Channel *channel,
 728               void **channel_ctx,
 729               const struct GNUNET_MessageHeader *message)
 730 {
 731   long n = (long) cls;
 732   struct MeshPeer *peer;
 733   struct MeshPingMessage *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_MESH_receive_done (channel);
 740   peer = &peers[n];
 741
 742   msg = (struct MeshPingMessage *) 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_MESH_MessageHandler handlers[] = {
 765   {&ping_handler, PING, sizeof (struct MeshPingMessage)},
 766   {&pong_handler, PONG, sizeof (struct MeshPingMessage)},
 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_MESH_Channel *channel,
 785                  const struct GNUNET_PeerIdentity *initiator,
 786                  uint32_t port, enum GNUNET_MESH_ChannelOption options)
 787 {
 788   long n = (long) cls;
 789   struct MeshPeer *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     test_task = GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_SECONDS,
 801                                               &start_test, NULL);
 802     return NULL;
 803   }
 804   GNUNET_assert (peer == peers[n].incoming);
 805   GNUNET_assert (peer->dest == &peers[n]);
 806   GNUNET_log (GNUNET_ERROR_TYPE_INFO, "%u <= %u %p\n",
 807               n, get_index (peer), channel);
 808   peers[n].incoming_ch = channel;
 809
 810   return NULL;
 811 }
 812
 813 /**
 814  * Function called whenever an inbound channel is destroyed.  Should clean up
 815  * any associated state.
 816  *
 817  * @param cls closure (set from GNUNET_MESH_connect)
 818  * @param channel connection to the other end (henceforth invalid)
 819  * @param channel_ctx place where local state associated
 820  *                   with the channel is stored
 821  */
 822 static void
 823 channel_cleaner (void *cls, const struct GNUNET_MESH_Channel *channel,
 824                  void *channel_ctx)
 825 {
 826   long n = (long) cls;
 827   struct MeshPeer *peer = &peers[n];
 828
 829   GNUNET_log (GNUNET_ERROR_TYPE_INFO,
 830               "Channel %p disconnected at peer %ld\n", channel, n);
 831   if (peer->ch == channel)
 832     peer->ch = NULL;
 833 }
 834
 835
 836 /**
 837  * Select a random peer that has no incoming channel
 838  *
 839  * @param peer ID of the peer connecting. NULL if irrelevant (warmup).
 840  *
 841  * @return Random peer not yet connected to.
 842  */
 843 static struct MeshPeer *
 844 select_random_peer (struct MeshPeer *peer)
 845 {
 846   unsigned int r;
 847
 848   do
 849   {
 850     r = GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK, peers_total);
 851   } while (NULL != peers[r].incoming);
 852   peers[r].incoming = peer;
 853
 854   return &peers[r];
 855 }
 856
 857 /**
 858  * START THE TEST ITSELF, AS WE ARE CONNECTED TO THE MESH SERVICES.
 859  *
 860  * Testcase continues when the root receives confirmation of connected peers,
 861  * on callback funtion ch.
 862  *
 863  * @param cls Closure (unsued).
 864  * @param tc Task Context.
 865  */
 866 static void
 867 start_test (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
 868 {
 869   enum GNUNET_MESH_ChannelOption flags;
 870   unsigned long i;
 871
 872   if ((GNUNET_SCHEDULER_REASON_SHUTDOWN & tc->reason) != 0)
 873     return;
 874
 875   GNUNET_log (GNUNET_ERROR_TYPE_INFO, "Start profiler\n");
 876
 877   flags = GNUNET_MESH_OPTION_DEFAULT;
 878   for (i = 0; i < peers_pinging; i++)
 879   {
 880     peers[i].dest = select_random_peer (&peers[i]);
 881     peers[i].ch = GNUNET_MESH_channel_create (peers[i].mesh, NULL,
 882                                               &peers[i].dest->id,
 883                                               1, flags);
 884     if (NULL == peers[i].ch)
 885     {
 886       GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "Channel %lu failed\n", i);
 887       GNUNET_MESH_TEST_cleanup (test_ctx);
 888       return;
 889     }
 890     GNUNET_log (GNUNET_ERROR_TYPE_INFO, "%u => %u %p\n",
 891                 i, get_index (peers[i].dest), peers[i].ch);
 892     peers[i].ping_task = GNUNET_SCHEDULER_add_delayed (delay_ms_rnd (2000),
 893                                                        &ping, &peers[i]);
 894   }
 895   peers_running = peers_total;
 896   if (GNUNET_SCHEDULER_NO_TASK != disconnect_task)
 897     GNUNET_SCHEDULER_cancel (disconnect_task);
 898   disconnect_task =
 899     GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_relative_multiply(round_time,
 900                                                                 number_rounds + 1),
 901                                   &disconnect_mesh_peers,
 902                                   (void *) __LINE__);
 903   GNUNET_SCHEDULER_add_delayed (round_time, &next_rnd, NULL);
 904 }
 905
 906
 907 /**
 908  * Do warmup: create some channels to spread information about the topology.
 909  */
 910 static void
 911 warmup (void)
 912 {
 913   struct MeshPeer *peer;
 914   unsigned int i;
 915
 916   for (i = 0; i < peers_total; i++)
 917   {
 918     peer = select_random_peer (NULL);
 919     GNUNET_log (GNUNET_ERROR_TYPE_INFO, "WARMUP %u => %u\n",
 920                 i, get_index (peer));
 921     peers[i].warmup_ch =
 922       GNUNET_MESH_channel_create (peers[i].mesh, NULL, &peer->id,
 923                                   1, GNUNET_MESH_OPTION_DEFAULT);
 924     if (NULL == peers[i].warmup_ch)
 925     {
 926       GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "Warmup %u failed\n", i);
 927       GNUNET_MESH_TEST_cleanup (test_ctx);
 928       return;
 929     }
 930   }
 931 }
 932
 933 /**
 934  * Callback to be called when the requested peer information is available
 935  *
 936  * @param cls the closure from GNUNET_TESTBED_peer_get_information()
 937  * @param op the operation this callback corresponds to
 938  * @param pinfo the result; will be NULL if the operation has failed
 939  * @param emsg error message if the operation has failed;
 940  *             NULL if the operation is successfull
 941  */
 942 static void
 943 peer_id_cb (void *cls,
 944        struct GNUNET_TESTBED_Operation *op,
 945        const struct GNUNET_TESTBED_PeerInformation *pinfo,
 946        const char *emsg)
 947 {
 948   long n = (long) cls;
 949
 950   if (NULL == pinfo || NULL != emsg)
 951   {
 952     GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "pi_cb: %s\n", emsg);
 953     abort_test (__LINE__);
 954     return;
 955   }
 956   peers[n].id = *(pinfo->result.id);
 957   GNUNET_log (GNUNET_ERROR_TYPE_INFO, " %u  id: %s\n",
 958               n, GNUNET_i2s (&peers[n].id));
 959   GNUNET_break (GNUNET_OK ==
 960                 GNUNET_CONTAINER_multipeermap_put (ids, &peers[n].id, &peers[n],
 961                                                    GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_FAST));
 962
 963   GNUNET_TESTBED_operation_done (peers[n].op);
 964   peers[n].op = NULL;
 965
 966   p_ids++;
 967   if (p_ids < peers_total)
 968     return;
 969   GNUNET_log (GNUNET_ERROR_TYPE_INFO, "Got all IDs, starting profiler\n");
 970   if (do_warmup)
 971   {
 972     warmup();
 973     return; /* start_test from incoming_channel */
 974   }
 975   test_task = GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_SECONDS,
 976                                             &start_test, NULL);
 977 }
 978
 979 /**
 980  * test main: start test when all peers are connected
 981  *
 982  * @param cls Closure.
 983  * @param ctx Argument to give to GNUNET_MESH_TEST_cleanup on test end.
 984  * @param num_peers Number of peers that are running.
 985  * @param testbed_peers Array of peers.
 986  * @param meshes Handle to each of the MESHs of the peers.
 987  */
 988 static void
 989 tmain (void *cls,
 990        struct GNUNET_MESH_TEST_Context *ctx,
 991        unsigned int num_peers,
 992        struct GNUNET_TESTBED_Peer **testbed_peers,
 993        struct GNUNET_MESH_Handle **meshes)
 994 {
 995   unsigned long i;
 996
 997   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "test main\n");
 998   test_ctx = ctx;
 999   GNUNET_assert (peers_total == num_peers);
1000   peers_running = num_peers;
1001   testbed_handles = testbed_peers;
1002   disconnect_task = GNUNET_SCHEDULER_add_delayed (SHORT_TIME,
1003                                                   &disconnect_mesh_peers,
1004                                                   (void *) __LINE__);
1005   shutdown_handle = GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_FOREVER_REL,
1006                                                   &shutdown_task, NULL);
1007   for (i = 0; i < peers_total; i++)
1008   {
1009     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "requesting id %ld\n", i);
1010     peers[i].up = GNUNET_YES;
1011     peers[i].mesh = meshes[i];
1012     peers[i].op =
1013       GNUNET_TESTBED_peer_get_information (testbed_handles[i],
1014                                            GNUNET_TESTBED_PIT_IDENTITY,
1015                                            &peer_id_cb, (void *) i);
1016   }
1017   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "requested peer ids\n");
1018   /* Continues from pi_cb -> do_test */
1019 }
1020
1021
1022 /**
1023  * Main: start profiler.
1024  */
1025 int
1026 main (int argc, char *argv[])
1027 {
1028   static uint32_t ports[2];
1029   const char *config_file;
1030
1031   config_file = ".profiler.conf";
1032
1033   if (4 > argc)
1034   {
1035     fprintf (stderr, "usage: %s ROUND_TIME PEERS PINGS [DO_WARMUP]\n", argv[0]);
1036     fprintf (stderr, "example: %s 30s 16 1 Y\n", argv[0]);
1037     return 1;
1038   }
1039
1040   if (GNUNET_OK != GNUNET_STRINGS_fancy_time_to_relative (argv[1], &round_time))
1041   {
1042     fprintf (stderr, "%s is not a valid time\n", argv[1]);
1043     return 1;
1044   }
1045
1046   peers_total = atoll (argv[2]);
1047   if (2 > peers_total)
1048   {
1049     fprintf (stderr, "%s peers is not valid (> 2)\n", argv[1]);
1050     return 1;
1051   }
1052   peers = GNUNET_malloc (sizeof (struct MeshPeer) * peers_total);
1053
1054   peers_pinging = atoll (argv[3]);
1055
1056   if (peers_total < 2 * peers_pinging)
1057   {
1058     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
1059                 "not enough peers, total should be > 2 * peers_pinging\n");
1060     return 1;
1061   }
1062
1063   do_warmup = (5 > argc || argv[4][0] != 'N');
1064
1065   ids = GNUNET_CONTAINER_multipeermap_create (2 * peers_total, GNUNET_YES);
1066   GNUNET_assert (NULL != ids);
1067   p_ids = 0;
1068   test_finished = GNUNET_NO;
1069   ports[0] = 1;
1070   ports[1] = 0;
1071   GNUNET_MESH_TEST_run ("mesh-profiler", config_file, peers_total,
1072                         &tmain, NULL, /* tmain cls */
1073                         &incoming_channel, &channel_cleaner,
1074                         handlers, ports);
1075   GNUNET_free (peers);
1076
1077   return 0;
1078 }
1079
1080 /* end of gnunet-mesh-profiler.c */
1081