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