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