src/dht/gnunet_dht_profiler.c

   1 /*
   2      This file is part of GNUnet.
   3      (C) 2014 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 /**
  22  * @file dht/gnunet_dht_profiler.c
  23  * @brief Profiler for GNUnet DHT
  24  * @author Sree Harsha Totakura <sreeharsha@totakura.in>
  25  */
  26
  27 #include "platform.h"
  28 #include "gnunet_util_lib.h"
  29 #include "gnunet_testbed_service.h"
  30 #include "gnunet_dht_service.h"
  31
  32 #define INFO(...)                                       \
  33   GNUNET_log (GNUNET_ERROR_TYPE_INFO, __VA_ARGS__)
  34
  35 #define DEBUG(...)                                           \
  36   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, __VA_ARGS__)
  37
  38 /**
  39  * Number of peers which should perform a PUT out of 100 peers
  40  */
  41 #define PUT_PROBABILITY 50
  42
  43 /**
  44  * Configuration
  45  */
  46 static struct GNUNET_CONFIGURATION_Handle *cfg;
  47
  48 /**
  49  * Name of the file with the hosts to run the test over
  50  */
  51 static char *hosts_file;
  52
  53 /**
  54  * Context for a peer which actively does DHT PUT/GET
  55  */
  56 struct ActiveContext;
  57
  58 /**
  59  * Context to hold data of peer
  60  */
  61 struct Context
  62 {
  63
  64   /**
  65    * The testbed peer this context belongs to
  66    */
  67   struct GNUNET_TESTBED_Peer *peer;
  68
  69   /**
  70    * Testbed operation acting on this peer
  71    */
  72   struct GNUNET_TESTBED_Operation *op;
  73
  74   /**
  75    * Active context; NULL if this peer is not an active peer
  76    */
  77   struct ActiveContext *ac;
  78 };
  79
  80
  81 /**
  82  * Context for a peer which actively does DHT PUT/GET
  83  */
  84 struct ActiveContext
  85 {
  86   /**
  87    * The linked peer context
  88    */
  89   struct Context *ctx;
  90
  91   /**
  92    * Handler to the DHT service
  93    */
  94   struct GNUNET_DHT_Handle *dht;
  95
  96   /**
  97    * The data used for do a PUT.  Will be NULL if a PUT hasn't been performed yet
  98    */
  99   void *put_data;
 100
 101   /**
 102    * The active context used for our DHT GET
 103    */
 104   struct ActiveContext *get_ac;
 105
 106   /**
 107    * The put handle
 108    */
 109   struct GNUNET_DHT_PutHandle *dht_put;
 110
 111   /**
 112    * The get handle
 113    */
 114   struct GNUNET_DHT_GetHandle *dht_get;
 115
 116   /**
 117    * The hash of the @e put_data
 118    */
 119   struct GNUNET_HashCode hash;
 120
 121   /**
 122    * Delay task
 123    */
 124   GNUNET_SCHEDULER_TaskIdentifier delay_task;
 125
 126   /**
 127    * The size of the @e put_data
 128    */
 129   uint16_t put_data_size;
 130
 131   /**
 132    * The number of peers currently doing GET on our data
 133    */
 134   uint16_t nrefs;
 135 };
 136
 137
 138 /**
 139  * An array of contexts.  The size of this array should be equal to @a num_peers
 140  */
 141 static struct Context *a_ctx;
 142
 143 /**
 144  * Array of active peers
 145  */
 146 static struct ActiveContext *a_ac;
 147
 148 /**
 149  * The delay between starting to do PUTS and GETS
 150  */
 151 static struct GNUNET_TIME_Relative delay;
 152
 153 /**
 154  * The timeout for GET and PUT
 155  */
 156 static struct GNUNET_TIME_Relative timeout;
 157
 158 /**
 159  * Number of peers
 160  */
 161 static unsigned int num_peers;
 162
 163 /**
 164  * Number of active peers
 165  */
 166 static unsigned int n_active;
 167
 168 /**
 169  * Number of DHT service connections we currently have
 170  */
 171 static unsigned int n_dht;
 172
 173 /**
 174  * Number of DHT PUTs made
 175  */
 176 static unsigned int n_puts;
 177
 178 /**
 179  * Number of DHT PUTs succeeded
 180  */
 181 static unsigned int n_puts_ok;
 182
 183 /**
 184  * Number of DHT PUTs failed
 185  */
 186 static unsigned int n_puts_fail;
 187
 188 /**
 189  * Number of DHT GETs made
 190  */
 191 static unsigned int n_gets;
 192
 193 /**
 194  * Number of DHT GETs succeeded
 195  */
 196 static unsigned int n_gets_ok;
 197
 198 /**
 199  * Number of DHT GETs succeeded
 200  */
 201 static unsigned int n_gets_fail;
 202
 203 /**
 204  * Replication degree
 205  */
 206 static unsigned int replication;
 207
 208 /**
 209  * Number of times we try to find the successor circle formation
 210  */
 211 static unsigned int max_searches;
 212
 213 /**
 214  * Testbed Operation (to get stats).
 215  */
 216 static struct GNUNET_TESTBED_Operation *bandwidth_stats_op;
 217
 218 /**
 219  * To get successor stats.
 220  */
 221 static struct GNUNET_TESTBED_Operation *successor_stats_op;
 222
 223 /**
 224  * Testbed peer handles.
 225  */
 226 static struct GNUNET_TESTBED_Peer **testbed_handles;
 227
 228 /**
 229  * Total number of messages sent by peer.
 230  */
 231 static uint64_t outgoing_bandwidth;
 232
 233 /**
 234  * Total number of messages received by peer.
 235  */
 236 static uint64_t incoming_bandwidth;
 237
 238 /**
 239  * Average number of hops taken to do put.
 240  */
 241 static unsigned int average_put_path_length;
 242
 243 /**
 244  * Average number of hops taken to do get.
 245  */
 246 static unsigned int average_get_path_length;
 247
 248 /**
 249  * Total put path length across all peers.
 250  */
 251 static unsigned int total_put_path_length;
 252
 253 /**
 254  * Total get path length across all peers.
 255  */
 256 static unsigned int total_get_path_length;
 257
 258 /**
 259  * Hashmap to store pair of peer and its corresponding successor.
 260  */
 261 static struct GNUNET_CONTAINER_MultiHashMap *successor_peer_hashmap;
 262
 263 /**
 264  * Key to start the lookup on successor_peer_hashmap.
 265  */
 266 static struct GNUNET_HashCode *start_key;
 267
 268 /**
 269  * Flag used to get the start_key.
 270  */
 271 static int flag = 0;
 272
 273 /**
 274  * Task to collect peer and its current successor statistics.
 275  */
 276 static GNUNET_SCHEDULER_TaskIdentifier successor_stats_task;
 277
 278 /**
 279  * Closure for successor_stats_task.
 280  */
 281 struct Collect_Stat_Context
 282 {
 283   /**
 284    * Current Peer Context.
 285    */
 286   struct Context *service_connect_ctx;
 287
 288   /**
 289    * Testbed operation acting on this peer
 290    */
 291   struct GNUNET_TESTBED_Operation *op;
 292 };
 293
 294 /**
 295  * List of all the peers contexts.
 296  */
 297 struct Context **peer_contexts = NULL;
 298
 299 /**
 300  * Counter to keep track of peers added to peer_context lists.
 301  */
 302 static int peers_started = 0;
 303
 304 /**
 305  * Task that collects successor statistics from all the peers.
 306  * @param cls
 307  * @param tc
 308  */
 309 static void
 310 collect_stats (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc);
 311
 312
 313 /**
 314  * Shutdown task.  Cleanup all resources and operations.
 315  *
 316  * @param cls NULL
 317  * @param tc scheduler task context
 318  */
 319 static void
 320 do_shutdown (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
 321 {
 322   struct ActiveContext *ac;
 323   unsigned int cnt;
 324
 325   if (NULL != a_ctx)
 326   {
 327     for (cnt=0; cnt < num_peers; cnt++)
 328     {
 329       if (NULL != a_ctx[cnt].op)
 330         GNUNET_TESTBED_operation_done (a_ctx[cnt].op);
 331
 332       /* Cleanup active context if this peer is an active peer */
 333       ac = a_ctx[cnt].ac;
 334       if (NULL == ac)
 335         continue;
 336       if (GNUNET_SCHEDULER_NO_TASK != ac->delay_task)
 337         GNUNET_SCHEDULER_cancel (ac->delay_task);
 338       if (NULL != ac->put_data)
 339         GNUNET_free (ac->put_data);
 340       if (NULL != ac->dht_put)
 341         GNUNET_DHT_put_cancel (ac->dht_put);
 342       if (NULL != ac->dht_get)
 343         GNUNET_DHT_get_stop (ac->dht_get);
 344     }
 345     GNUNET_free (a_ctx);
 346     a_ctx = NULL;
 347   }
 348   if(NULL != bandwidth_stats_op)
 349     GNUNET_TESTBED_operation_done (bandwidth_stats_op);
 350   bandwidth_stats_op = NULL;
 351   GNUNET_free_non_null (a_ac);
 352 }
 353
 354
 355 /**
 356  * Stats callback. Finish the stats testbed operation and when all stats have
 357  * been iterated, shutdown the test.
 358  *
 359  * @param cls closure
 360  * @param op the operation that has been finished
 361  * @param emsg error message in case the operation has failed; will be NULL if
 362  *          operation has executed successfully.
 363  */
 364 static void
 365 bandwidth_stats_cont (void *cls,
 366                       struct GNUNET_TESTBED_Operation *op,
 367                       const char *emsg)
 368 {
 369   INFO ("# Outgoing bandwidth: %u\n", outgoing_bandwidth);
 370   INFO ("# Incoming bandwidth: %u\n", incoming_bandwidth);
 371   GNUNET_SCHEDULER_shutdown ();
 372 }
 373
 374
 375 /**
 376  * Process statistic values.
 377  *
 378  * @param cls closure
 379  * @param peer the peer the statistic belong to
 380  * @param subsystem name of subsystem that created the statistic
 381  * @param name the name of the datum
 382  * @param value the current value
 383  * @param is_persistent GNUNET_YES if the value is persistent, GNUNET_NO if not
 384  * @return GNUNET_OK to continue, GNUNET_SYSERR to abort iteration
 385  */
 386 static int
 387 bandwidth_stats_iterator (void *cls,
 388                           const struct GNUNET_TESTBED_Peer *peer,
 389                           const char *subsystem,
 390                           const char *name,
 391                           uint64_t value,
 392                           int is_persistent)
 393 {
 394    static const char *s_sent = "# Bytes transmitted to other peers";
 395    static const char *s_recv = "# Bytes received from other peers";
 396
 397    if (0 == strncmp (s_sent, name, strlen (s_sent)))
 398      outgoing_bandwidth = outgoing_bandwidth + value;
 399    else if (0 == strncmp(s_recv, name, strlen (s_recv)))
 400      incoming_bandwidth = incoming_bandwidth + value;
 401    else
 402      return GNUNET_OK;
 403    DEBUG ("Bandwith - Out: %lu; In: %lu\n",
 404           (unsigned long) outgoing_bandwidth,
 405           (unsigned long) incoming_bandwidth);
 406    return GNUNET_OK;
 407 }
 408
 409
 410 static void
 411 summarize ()
 412 {
 413   INFO ("# PUTS made: %u\n", n_puts);
 414   INFO ("# PUTS succeeded: %u\n", n_puts_ok);
 415   INFO ("# PUTS failed: %u\n", n_puts_fail);
 416   INFO ("# GETS made: %u\n", n_gets);
 417   INFO ("# GETS succeeded: %u\n", n_gets_ok);
 418   INFO ("# GETS failed: %u\n", n_gets_fail);
 419   INFO ("# average_put_path_length: %u\n", average_put_path_length);
 420   INFO ("# average_get_path_length: %u\n", average_get_path_length);
 421
 422   if (NULL == testbed_handles)
 423   {
 424     INFO ("No peers found\n");
 425     return;
 426   }
 427   /* Collect Stats*/
 428   bandwidth_stats_op = GNUNET_TESTBED_get_statistics (n_active, testbed_handles,
 429                                                       "dht", NULL,
 430                                                        bandwidth_stats_iterator,
 431                                                        bandwidth_stats_cont, NULL);
 432 }
 433
 434
 435 /**
 436  * Task to cancel DHT GET.
 437  *
 438  * @param cls NULL
 439  * @param tc scheduler task context
 440  */
 441 static void
 442 cancel_get (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
 443 {
 444   struct ActiveContext *ac = cls;
 445
 446   ac->delay_task = GNUNET_SCHEDULER_NO_TASK;
 447   GNUNET_assert (NULL != ac->dht_get);
 448   GNUNET_DHT_get_stop (ac->dht_get);
 449   ac->dht_get = NULL;
 450   n_gets_fail++;
 451
 452   /* If profiling is complete, summarize */
 453   if (n_active == n_gets_fail + n_gets_ok)
 454     summarize ();
 455
 456 }
 457
 458
 459 /**
 460  * Iterator called on each result obtained for a DHT
 461  * operation that expects a reply
 462  *
 463  * @param cls closure
 464  * @param exp when will this value expire
 465  * @param key key of the result
 466  * @param get_path peers on reply path (or NULL if not recorded)
 467  *                 [0] = datastore's first neighbor, [length - 1] = local peer
 468  * @param get_path_length number of entries in @a get_path
 469  * @param put_path peers on the PUT path (or NULL if not recorded)
 470  *                 [0] = origin, [length - 1] = datastore
 471  * @param put_path_length number of entries in @a put_path
 472  * @param type type of the result
 473  * @param size number of bytes in @a data
 474  * @param data pointer to the result data
 475  */
 476 static void
 477 get_iter (void *cls,
 478           struct GNUNET_TIME_Absolute exp,
 479           const struct GNUNET_HashCode *key,
 480           const struct GNUNET_PeerIdentity *get_path,
 481           unsigned int get_path_length,
 482           const struct GNUNET_PeerIdentity *put_path,
 483           unsigned int put_path_length,
 484           enum GNUNET_BLOCK_Type type,
 485           size_t size, const void *data)
 486 {
 487   struct ActiveContext *ac = cls;
 488   struct ActiveContext *get_ac = ac->get_ac;
 489
 490   /* Check the keys of put and get match or not. */
 491   GNUNET_assert (0 == memcmp (key, &get_ac->hash, sizeof (struct GNUNET_HashCode)));
 492   /* we found the data we are looking for */
 493   DEBUG ("We found a GET request; %u remaining\n", n_gets - (n_gets_fail + n_gets_ok));
 494   n_gets_ok++;
 495   get_ac->nrefs--;
 496   GNUNET_DHT_get_stop (ac->dht_get);
 497   ac->dht_get = NULL;
 498   GNUNET_SCHEDULER_cancel (ac->delay_task);
 499   ac->delay_task = GNUNET_SCHEDULER_NO_TASK;
 500
 501   total_put_path_length = total_put_path_length + put_path_length;
 502   total_get_path_length = total_get_path_length + get_path_length;
 503
 504   /* Summarize if profiling is complete */
 505   if (n_active == n_gets_fail + n_gets_ok)
 506   {
 507     average_put_path_length = total_put_path_length/n_active;
 508     average_get_path_length = total_get_path_length/n_active;
 509     summarize ();
 510   }
 511 }
 512
 513
 514 /**
 515  * Task to do DHT GETs
 516  *
 517  * @param cls the active context
 518  * @param tc the scheduler task context
 519  */
 520 static void
 521 delayed_get (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
 522 {
 523   struct ActiveContext *ac = cls;
 524   struct ActiveContext *get_ac;
 525   unsigned int r;
 526
 527   ac->delay_task = GNUNET_SCHEDULER_NO_TASK;
 528   get_ac = NULL;
 529   while (1)
 530   {
 531     r = GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK, n_active);
 532     get_ac = &a_ac[r];
 533     if (NULL != get_ac->put_data)
 534       break;
 535   }
 536   get_ac->nrefs++;
 537   ac->get_ac = get_ac;
 538   DEBUG ("Doing a DHT GET for data of size %u\n", get_ac->put_data_size);
 539   ac->dht_get = GNUNET_DHT_get_start (ac->dht,
 540                                       GNUNET_BLOCK_TYPE_TEST,
 541                                       &get_ac->hash,
 542                                       1, /* replication level */
 543                                       GNUNET_DHT_RO_NONE,
 544                                       NULL, 0, /* extended query and size */
 545                                       get_iter, ac); /* GET iterator and closure
 546                                                         */
 547   n_gets++;
 548
 549   /* schedule the timeout task for GET */
 550   ac->delay_task = GNUNET_SCHEDULER_add_delayed (timeout, &cancel_get, ac);
 551 }
 552
 553
 554 /**
 555  * Queue up a delayed task for doing DHT GET
 556  *
 557  * @param cls the active context
 558  * @param success #GNUNET_OK if the PUT was transmitted,
 559  *                #GNUNET_NO on timeout,
 560  *                #GNUNET_SYSERR on disconnect from service
 561  *                after the PUT message was transmitted
 562  *                (so we don't know if it was received or not)
 563  */
 564 static void
 565 put_cont (void *cls, int success)
 566 {
 567   struct ActiveContext *ac = cls;
 568
 569   ac->dht_put = NULL;
 570   if (success)
 571     n_puts_ok++;
 572   else
 573     n_puts_fail++;
 574   ac->delay_task = GNUNET_SCHEDULER_add_delayed (delay, &delayed_get, ac);
 575 }
 576
 577
 578 /**
 579  * Task to do DHT PUTS
 580  *
 581  * @param cls the active context
 582  * @param tc the scheduler task context
 583  */
 584 static void
 585 delayed_put (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
 586 {
 587   struct ActiveContext *ac = cls;
 588   DEBUG("PUT SUPU \n");
 589   ac->delay_task = GNUNET_SCHEDULER_NO_TASK;
 590   /* Generate and DHT PUT some random data */
 591   ac->put_data_size = 16;       /* minimum */
 592   ac->put_data_size += GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
 593                                                  (63*1024));
 594   ac->put_data = GNUNET_malloc (ac->put_data_size);
 595   GNUNET_CRYPTO_random_block (GNUNET_CRYPTO_QUALITY_WEAK,
 596                               ac->put_data, ac->put_data_size);
 597   GNUNET_CRYPTO_hash (ac->put_data, ac->put_data_size, &ac->hash);
 598   DEBUG ("Doing a DHT PUT with data of size %u\n", ac->put_data_size);
 599   ac->dht_put = GNUNET_DHT_put (ac->dht, &ac->hash,
 600                                 replication,
 601                                 GNUNET_DHT_RO_NONE,
 602                                 GNUNET_BLOCK_TYPE_TEST,
 603                                 ac->put_data_size,
 604                                 ac->put_data,
 605                                 GNUNET_TIME_UNIT_FOREVER_ABS, /* expiration time */
 606                                 timeout,                      /* PUT timeout */
 607                                 put_cont, ac);                /* continuation and its closure */
 608   n_puts++;
 609 }
 610
 611
 612 /**
 613  * Connection to DHT has been established.  Call the delay task.
 614  *
 615  * @param cls the active context
 616  * @param op the operation that has been finished
 617  * @param ca_result the service handle returned from GNUNET_TESTBED_ConnectAdapter()
 618  * @param emsg error message in case the operation has failed; will be NULL if
 619  *          operation has executed successfully.
 620  */
 621 static void
 622 dht_connected (void *cls,
 623                struct GNUNET_TESTBED_Operation *op,
 624                void *ca_result,
 625                const char *emsg)
 626 {
 627   struct ActiveContext *ac = cls;
 628   struct Context *ctx = ac->ctx;
 629
 630   GNUNET_assert (NULL != ctx);
 631   GNUNET_assert (NULL != ctx->op);
 632   GNUNET_assert (ctx->op == op);
 633   ac->dht = (struct GNUNET_DHT_Handle *) ca_result;
 634   if (NULL != emsg)
 635   {
 636     GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "Connection to DHT service failed: %s\n", emsg);
 637     GNUNET_TESTBED_operation_done (ctx->op); /* Calls dht_disconnect() */
 638     ctx->op = NULL;
 639     return;
 640   }
 641
 642   ac->delay_task = GNUNET_SCHEDULER_add_delayed (delay, &delayed_put, ac);
 643 }
 644
 645
 646 /**
 647  * Connect to DHT service and return the DHT client handler
 648  *
 649  * @param cls the active context
 650  * @param cfg configuration of the peer to connect to; will be available until
 651  *          GNUNET_TESTBED_operation_done() is called on the operation returned
 652  *          from GNUNET_TESTBED_service_connect()
 653  * @return service handle to return in 'op_result', NULL on error
 654  */
 655 static void *
 656 dht_connect (void *cls, const struct GNUNET_CONFIGURATION_Handle *cfg)
 657 {
 658   n_dht++;
 659   return GNUNET_DHT_connect (cfg, 10);
 660 }
 661
 662
 663 /**
 664  * Adapter function called to destroy a connection to
 665  * a service.
 666  *
 667  * @param cls the active context
 668  * @param op_result service handle returned from the connect adapter
 669  */
 670 static void
 671 dht_disconnect (void *cls, void *op_result)
 672 {
 673   struct ActiveContext *ac = cls;
 674
 675   GNUNET_assert (NULL != ac->dht);
 676   GNUNET_assert (ac->dht == op_result);
 677   GNUNET_DHT_disconnect (ac->dht);
 678   n_dht--;
 679   if (0 == n_dht)
 680     GNUNET_SCHEDULER_shutdown ();
 681 }
 682
 683
 684 /**
 685  * FIXME:Verify where is n_active used. Should this service be started only
 686  * for n_active peers?
 687  * Start testbed service for all the peers.
 688  */
 689 static void
 690 start_testbed_service_on_all_peers()
 691 {
 692   unsigned int i;
 693   for(i = 0; i < peers_started; i++)
 694   {
 695       struct Context *ctx = peer_contexts[i];
 696       DEBUG("GNUNET_TESTBED_service_connect \n");
 697       ctx->op =
 698               GNUNET_TESTBED_service_connect (ctx,
 699                                               ctx->peer,
 700                                               "dht",
 701                                               &dht_connected, ctx->ac,
 702                                               &dht_connect,
 703                                               &dht_disconnect,
 704                                               ctx->ac);
 705
 706   }
 707 }
 708
 709
 710 /**
 711  * Stats callback. Iterate over the hashmap and check if all th peers form
 712  * a virtual ring topology.
 713  *
 714  * @param cls closure
 715  * @param op the operation that has been finished
 716  * @param emsg error message in case the operation has failed; will be NULL if
 717  *          operation has executed successfully.
 718  */
 719 static void
 720 successor_stats_cont (void *cls,
 721                       struct GNUNET_TESTBED_Operation *op,
 722                       const char *emsg)
 723 {
 724   struct GNUNET_HashCode *val;
 725   struct GNUNET_HashCode *start_val;
 726   int count = 0;
 727   struct GNUNET_HashCode *key;
 728
 729   start_val =(struct GNUNET_HashCode *) GNUNET_CONTAINER_multihashmap_get(successor_peer_hashmap,
 730                                                 start_key);
 731
 732   val = GNUNET_new(struct GNUNET_HashCode);
 733   val = start_val;
 734   while (count < num_peers)
 735   {
 736     key = GNUNET_new(struct GNUNET_HashCode);
 737     key = val;
 738     val = GNUNET_CONTAINER_multihashmap_get (successor_peer_hashmap,
 739                                              key);
 740     GNUNET_assert(NULL != val);
 741     count++;
 742   }
 743
 744   if (start_val == val)
 745   {
 746     DEBUG("Circle completed\n");
 747     if (GNUNET_SCHEDULER_NO_TASK != successor_stats_task)
 748     {
 749       successor_stats_task = GNUNET_SCHEDULER_NO_TASK;
 750       //FIXME: free hashmap.
 751     }
 752     successor_stats_op = NULL;
 753
 754     if(GNUNET_SCHEDULER_NO_TASK == successor_stats_task)
 755     {
 756       start_testbed_service_on_all_peers();
 757     }
 758
 759     return;
 760   }
 761   else
 762   {
 763     static unsigned int tries;
 764
 765     DEBUG("Circle not complete\n");
 766     if (max_searches == ++tries)
 767     {
 768       GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 769                   "Maximum tries %u exceeded while checking successor TOTAL TRIES %u"
 770                   " cirle formation.  Exiting\n",
 771                   max_searches,tries);
 772       if (GNUNET_SCHEDULER_NO_TASK != successor_stats_task)
 773       {
 774         successor_stats_task = GNUNET_SCHEDULER_NO_TASK;
 775         //FIXME: free hashmap
 776       }
 777       successor_stats_op = NULL;
 778
 779       if(GNUNET_SCHEDULER_NO_TASK == successor_stats_task)
 780       {
 781         start_testbed_service_on_all_peers();
 782       }
 783
 784       return;
 785     }
 786
 787     //FIXME: change delay use exponential back off.
 788     successor_stats_task = GNUNET_SCHEDULER_add_delayed (delay, &collect_stats, cls);
 789   }
 790 }
 791
 792
 793 /**
 794  * Process successor statistic values.
 795  *
 796  * @param cls closure
 797  * @param peer the peer the statistic belong to
 798  * @param subsystem name of subsystem that created the statistic
 799  * @param name the name of the datum
 800  * @param value the current value
 801  * @param is_persistent GNUNET_YES if the value is persistent, GNUNET_NO if not
 802  * @return GNUNET_OK to continue, GNUNET_SYSERR to abort iteration
 803  */
 804 static int
 805 successor_stats_iterator (void *cls,
 806                           const struct GNUNET_TESTBED_Peer *peer,
 807                           const char *subsystem,
 808                           const char *name,
 809                           uint64_t value,
 810                           int is_persistent)
 811 {
 812   static const char *key_string = "XDHT";
 813
 814   if (0 == strncmp (key_string, name, strlen (key_string)))
 815   {
 816     char *my_id_str;
 817     char successor_str[13];
 818     char truncated_my_id_str[13];
 819     char truncated_successor_str[13];
 820     struct GNUNET_HashCode *my_id_key;
 821     struct GNUNET_HashCode *succ_key;
 822
 823     strtok((char *)name,":");
 824     my_id_str = strtok(NULL,":");
 825
 826     strncpy(truncated_my_id_str, my_id_str, 12);
 827     truncated_my_id_str[12] = '\0';
 828
 829     my_id_key = GNUNET_new(struct GNUNET_HashCode);
 830     GNUNET_CRYPTO_hash (truncated_my_id_str, sizeof(truncated_my_id_str),my_id_key);
 831
 832     GNUNET_STRINGS_data_to_string(&value, sizeof(uint64_t), successor_str, 13);
 833     strncpy(truncated_successor_str, successor_str, 12);
 834     truncated_successor_str[12] ='\0';
 835
 836     succ_key = GNUNET_new(struct GNUNET_HashCode);
 837     GNUNET_CRYPTO_hash (truncated_successor_str, sizeof(truncated_successor_str),succ_key);
 838
 839     if (0 == flag)
 840     {
 841       start_key = my_id_key;
 842       flag = 1;
 843     }
 844     GNUNET_CONTAINER_multihashmap_put (successor_peer_hashmap,
 845                                        my_id_key, (void *)succ_key,
 846                                        GNUNET_CONTAINER_MULTIHASHMAPOPTION_REPLACE);
 847   }
 848   return GNUNET_OK;
 849 }
 850
 851
 852 /*
 853  * Task that collects peer and its corresponding successors.
 854  *
 855  * @param cls Closure (NULL).
 856  * @param tc Task Context.
 857  */
 858 static void
 859 collect_stats (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
 860 {
 861   if ((GNUNET_SCHEDULER_REASON_SHUTDOWN & tc->reason) != 0)
 862     return;
 863
 864   GNUNET_log (GNUNET_ERROR_TYPE_INFO, "Start collecting statistics...\n");
 865
 866  /* Check for successor pointer, don't start put till the virtual ring topology
 867    is not created. */
 868   successor_stats_op =
 869           GNUNET_TESTBED_get_statistics (num_peers, testbed_handles,
 870                                          "dht", NULL,
 871                                           successor_stats_iterator,
 872                                           successor_stats_cont, cls);
 873 }
 874
 875 /**
 876  * Callback called when DHT service on the peer is started
 877  *
 878  * @param cls the context
 879  * @param op the operation that has been finished
 880  * @param emsg error message in case the operation has failed; will be NULL if
 881  *          operation has executed successfully.
 882  */
 883 static void
 884 service_started (void *cls,
 885                  struct GNUNET_TESTBED_Operation *op,
 886                  const char *emsg)
 887 {
 888   struct Context *ctx = cls;
 889
 890   GNUNET_assert (NULL != ctx);
 891   GNUNET_assert (NULL != ctx->op);
 892   GNUNET_TESTBED_operation_done (ctx->op);
 893   ctx->op = NULL;
 894   if (NULL == ctx->ac)
 895     return;
 896
 897   if (NULL == peer_contexts)
 898   {
 899     peer_contexts = GNUNET_malloc(num_peers * sizeof(struct Context *));
 900   }
 901
 902   peer_contexts[peers_started] = ctx;
 903   peers_started++;
 904   DEBUG("Peers Started = %d \n", peers_started);
 905
 906   if (GNUNET_SCHEDULER_NO_TASK == successor_stats_task)
 907   {
 908      DEBUG("successor_stats_task \n");
 909      struct Collect_Stat_Context *collect_stat_cls = GNUNET_new(struct Collect_Stat_Context);
 910      collect_stat_cls->service_connect_ctx = cls;
 911      collect_stat_cls->op = op;
 912
 913      successor_peer_hashmap = GNUNET_CONTAINER_multihashmap_create (num_peers,
 914                                                                     GNUNET_NO);
 915      successor_stats_task = GNUNET_SCHEDULER_add_delayed (delay,
 916                                                           &collect_stats,
 917                                                           collect_stat_cls);
 918   }
 919 }
 920
 921
 922 /**
 923  * Signature of a main function for a testcase.
 924  *
 925  * @param cls closure
 926  * @param h the run handle
 927  * @param num_peers number of peers in 'peers'
 928  * @param peers handle to peers run in the testbed
 929  * @param links_succeeded the number of overlay link connection attempts that
 930  *          succeeded
 931  * @param links_failed the number of overlay link
 932  */
 933 static void
 934 test_run (void *cls,
 935           struct GNUNET_TESTBED_RunHandle *h,
 936           unsigned int num_peers, struct GNUNET_TESTBED_Peer **peers,
 937           unsigned int links_succeeded,
 938           unsigned int links_failed)
 939 {
 940   unsigned int cnt;
 941   unsigned int ac_cnt;
 942
 943   testbed_handles = peers;
 944   if (NULL == peers)
 945   {
 946     /* exit */
 947     GNUNET_assert (0);
 948   }
 949   INFO ("%u peers started\n", num_peers);
 950   a_ctx = GNUNET_malloc (sizeof (struct Context) * num_peers);
 951
 952   /* select the peers which actively participate in profiling */
 953   n_active = num_peers * PUT_PROBABILITY / 100;
 954   if (0 == n_active)
 955   {
 956     GNUNET_SCHEDULER_shutdown ();
 957     GNUNET_free (a_ctx);
 958     return;
 959   }
 960
 961   a_ac = GNUNET_malloc (n_active * sizeof (struct ActiveContext));
 962   ac_cnt = 0;
 963   for (cnt = 0; cnt < num_peers && ac_cnt < n_active; cnt++)
 964   {
 965     if (GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK, 100) >=
 966         PUT_PROBABILITY)
 967       continue;
 968     a_ctx[cnt].ac = &a_ac[ac_cnt];
 969     a_ac[ac_cnt].ctx = &a_ctx[cnt];
 970     ac_cnt++;
 971   }
 972   n_active = ac_cnt;
 973   a_ac = GNUNET_realloc (a_ac, n_active * sizeof (struct ActiveContext));
 974   INFO ("Active peers: %u\n", n_active);
 975
 976   /* start DHT service on all peers */
 977   for (cnt = 0; cnt < num_peers; cnt++)
 978   {
 979     a_ctx[cnt].peer = peers[cnt];
 980     a_ctx[cnt].op = GNUNET_TESTBED_peer_manage_service (&a_ctx[cnt],
 981                                                         peers[cnt],
 982                                                         "dht",
 983                                                         &service_started,
 984                                                         &a_ctx[cnt],
 985                                                         1);
 986   }
 987 }
 988
 989
 990 /**
 991  * Main function that will be run by the scheduler.
 992  *
 993  * @param cls closure
 994  * @param args remaining command-line arguments
 995  * @param cfgfile name of the configuration file used (for saving, can be NULL!)
 996  * @param config configuration
 997  */
 998 static void
 999 run (void *cls, char *const *args, const char *cfgfile,
1000      const struct GNUNET_CONFIGURATION_Handle *config)
1001 {
1002   uint64_t event_mask;
1003
1004   if (0 == num_peers)
1005   {
1006     GNUNET_log (GNUNET_ERROR_TYPE_ERROR, _("Exiting as the number of peers is %u\n"),
1007                 num_peers);
1008     return;
1009   }
1010   cfg = GNUNET_CONFIGURATION_dup (config);
1011   event_mask = 0;
1012   GNUNET_TESTBED_run (hosts_file, cfg, num_peers, event_mask, NULL,
1013                       NULL, &test_run, NULL);
1014   GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_FOREVER_REL, &do_shutdown,
1015                                 NULL);
1016 }
1017
1018
1019 /**
1020  * Main function.
1021  *
1022  * @return 0 on success
1023  */
1024 int
1025 main (int argc, char *const *argv)
1026 {
1027   int rc;
1028
1029   static struct GNUNET_GETOPT_CommandLineOption options[] = {
1030     {'n', "peers", "COUNT",
1031      gettext_noop ("number of peers to start"),
1032      1, &GNUNET_GETOPT_set_uint, &num_peers},
1033     {'s', "searches", "COUNT",
1034      gettext_noop ("maximum number of times we try to search for successor circle formation (default is 1)"),
1035      1, &GNUNET_GETOPT_set_uint, &max_searches},
1036     {'H', "hosts", "FILENAME",
1037      gettext_noop ("name of the file with the login information for the testbed"),
1038      1, &GNUNET_GETOPT_set_string, &hosts_file},
1039     {'d', "delay", "DELAY",
1040      gettext_noop ("delay for starting DHT PUT and GET"),
1041      1, &GNUNET_GETOPT_set_relative_time, &delay},
1042     {'r', "replication", "DEGREE",
1043      gettext_noop ("replication degree for DHT PUTs"),
1044      1, &GNUNET_GETOPT_set_uint, &replication},
1045     {'t', "timeout", "TIMEOUT",
1046      gettext_noop ("timeout for DHT PUT and GET requests"),
1047      1, &GNUNET_GETOPT_set_relative_time, &timeout},
1048     GNUNET_GETOPT_OPTION_END
1049   };
1050
1051   max_searches = 10;
1052   if (GNUNET_OK != GNUNET_STRINGS_get_utf8_args (argc, argv, &argc, &argv))
1053     return 2;
1054   delay = GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 2); /* default delay */
1055   timeout = GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MINUTES, 1); /* default timeout */
1056   replication = 1;      /* default replication */
1057   rc = 0;
1058   if (GNUNET_OK !=
1059       GNUNET_PROGRAM_run (argc, argv, "dht-profiler",
1060                           gettext_noop
1061                           ("Measure quality and performance of the DHT service."),
1062                           options, &run, NULL))
1063     rc = 1;
1064   return rc;
1065 }