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