gnunet_dht_profiler.c

Go to the documentation of this file.

/*
     This file is part of GNUnet.
     Copyright (C) 2014, 2018 GNUnet e.V.

     GNUnet is free software: you can redistribute it and/or modify it
     under the terms of the GNU Affero General Public License as published
     by the Free Software Foundation, either version 3 of the License,
     or (at your option) any later version.

     GNUnet is distributed in the hope that it will be useful, but
     WITHOUT ANY WARRANTY; without even the implied warranty of
     MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
     Affero General Public License for more details.

     You should have received a copy of the GNU Affero General Public License
     along with this program.  If not, see <http://www.gnu.org/licenses/>.

     SPDX-License-Identifier: AGPL3.0-or-later
 */

#include "platform.h"
#include "gnunet_util_lib.h"
#include "gnunet_testbed_service.h"
#include "gnunet_dht_service.h"
#include "gnunet_constants.h"


#define MESSAGE(...)                                       \
  GNUNET_log (GNUNET_ERROR_TYPE_MESSAGE, __VA_ARGS__)

#define DEBUG(...)                                           \
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, __VA_ARGS__)

static unsigned int put_probability = 100;

static const struct GNUNET_CONFIGURATION_Handle *cfg;

static char *hosts_file;

struct ActiveContext;

struct Context
{
  struct GNUNET_TESTBED_Peer *peer;

  struct GNUNET_TESTBED_Operation *op;

  struct ActiveContext *ac;
};


struct ActiveContext
{
  struct Context *ctx;

  struct GNUNET_DHT_Handle *dht;

  struct ActiveContext *get_ac;

  struct GNUNET_DHT_PutHandle *dht_put;

  struct GNUNET_DHT_GetHandle *dht_get;

  struct GNUNET_HashCode *hash;

  struct GNUNET_SCHEDULER_Task *delay_task;

  unsigned int put_count;

  uint16_t nrefs;
};


static struct Context *a_ctx;

static struct ActiveContext *a_ac;

static struct GNUNET_TIME_Relative delay_stats;

static struct GNUNET_TIME_Relative delay_put;

static struct GNUNET_TIME_Relative delay_get;

static struct GNUNET_TIME_Relative timeout;

static unsigned int num_peers;

static unsigned int n_active;

static unsigned int n_dht;

static unsigned long long n_puts;

static unsigned int num_puts_per_peer = 1;

static unsigned long long n_puts_ok;

static unsigned int n_gets;

static unsigned int n_gets_ok;

static unsigned int n_gets_fail;

static unsigned int replication;

static struct GNUNET_TESTBED_Operation *bandwidth_stats_op;

static struct GNUNET_TESTBED_Peer **testbed_handles;

static uint64_t outgoing_bandwidth;

static uint64_t incoming_bandwidth;

static double average_put_path_length;

static double average_get_path_length;

static unsigned int total_put_path_length;

static unsigned int total_get_path_length;

static int peers_started = 0;

static enum
{
  MODE_PUT = 0,

  MODE_GET = 1
} mode;


static int in_shutdown = 0;


static void
start_profiling (void);


static void
do_shutdown (void *cls)
{
  struct ActiveContext *ac;

  in_shutdown = GNUNET_YES;
  if (NULL != a_ctx)
  {
    for (unsigned int cnt = 0; cnt < num_peers; cnt++)
    {
      /* Cleanup active context if this peer is an active peer */
      ac = a_ctx[cnt].ac;
      if (NULL != ac)
      {
        if (NULL != ac->delay_task)
          GNUNET_SCHEDULER_cancel (ac->delay_task);
        if (NULL != ac->hash)
          free (ac->hash);
        if (NULL != ac->dht_put)
          GNUNET_DHT_put_cancel (ac->dht_put);
        if (NULL != ac->dht_get)
          GNUNET_DHT_get_stop (ac->dht_get);
      }
      /* Cleanup testbed operation handle at the last as this operation may
         contain service connection to DHT */
      if (NULL != a_ctx[cnt].op)
        GNUNET_TESTBED_operation_done (a_ctx[cnt].op);
    }
    GNUNET_free (a_ctx);
    a_ctx = NULL;
  }
  // FIXME: Should we collect stats only for put/get not for other messages.
  if (NULL != bandwidth_stats_op)
  {
    GNUNET_TESTBED_operation_done (bandwidth_stats_op);
    bandwidth_stats_op = NULL;
  }
  GNUNET_free (a_ac);
}


static void
bandwidth_stats_cont (void *cls,
                      struct GNUNET_TESTBED_Operation *op,
                      const char *emsg)
{
  MESSAGE ("# Outgoing (core) bandwidth: %llu bytes\n",
           (unsigned long long) outgoing_bandwidth);
  MESSAGE ("# Incoming (core) bandwidth: %llu bytes\n",
           (unsigned long long) incoming_bandwidth);
  fprintf (stderr,
           "Benchmark done. Collect data via gnunet-statistics, then press ENTER to exit.\n");
  (void) getchar ();
  GNUNET_SCHEDULER_shutdown ();
}


static int
bandwidth_stats_iterator (void *cls,
                          const struct GNUNET_TESTBED_Peer *peer,
                          const char *subsystem,
                          const char *name,
                          uint64_t value,
                          int is_persistent)
{
  static const char *s_sent = "# bytes encrypted";
  static const char *s_recv = "# bytes decrypted";

  if (0 == strncmp (s_sent, name, strlen (s_sent)))
    outgoing_bandwidth = outgoing_bandwidth + value;
  else if (0 == strncmp (s_recv, name, strlen (s_recv)))
    incoming_bandwidth = incoming_bandwidth + value;
  return GNUNET_OK;
}


static void
summarize ()
{
  MESSAGE ("# PUTS started: %llu\n",
           n_puts);
  MESSAGE ("# PUTS succeeded: %llu\n",
           n_puts_ok);
  MESSAGE ("# GETS made: %u\n",
           n_gets);
  MESSAGE ("# GETS succeeded: %u\n",
           n_gets_ok);
  MESSAGE ("# GETS failed: %u\n",
           n_gets_fail);
  MESSAGE ("# average_put_path_length: %f\n",
           average_put_path_length);
  MESSAGE ("# average_get_path_length: %f\n",
           average_get_path_length);

  if (NULL == testbed_handles)
  {
    MESSAGE ("No peers found\n");
    return;
  }
  /* Collect Stats*/
  bandwidth_stats_op = GNUNET_TESTBED_get_statistics (n_active,
                                                      testbed_handles,
                                                      "core",
                                                      NULL,
                                                      &bandwidth_stats_iterator,
                                                      &bandwidth_stats_cont,
                                                      NULL);
}


static void
cancel_get (void *cls)
{
  struct ActiveContext *ac = cls;
  struct Context *ctx = ac->ctx;

  ac->delay_task = NULL;
  GNUNET_assert (NULL != ac->dht_get);
  GNUNET_DHT_get_stop (ac->dht_get);
  ac->dht_get = NULL;
  n_gets_fail++;
  GNUNET_assert (NULL != ctx->op);
  GNUNET_TESTBED_operation_done (ctx->op);
  ctx->op = NULL;

  /* If profiling is complete, summarize */
  if (n_active == n_gets_fail + n_gets_ok)
  {
    average_put_path_length = (double) total_put_path_length
                              / (double) n_active;
    average_get_path_length = (double) total_get_path_length
                              / (double ) n_gets_ok;
    summarize ();
  }
}


static void
get_iter (void *cls,
          struct GNUNET_TIME_Absolute exp,
          const struct GNUNET_HashCode *key,
          const struct GNUNET_PeerIdentity *get_path,
          unsigned int get_path_length,
          const struct GNUNET_PeerIdentity *put_path,
          unsigned int put_path_length,
          enum GNUNET_BLOCK_Type type,
          size_t size, const void *data)
{
  struct ActiveContext *ac = cls;
  struct ActiveContext *get_ac = ac->get_ac;
  struct Context *ctx = ac->ctx;

  /* we found the data we are looking for */
  DEBUG ("We found a GET request; %u remaining\n",
         n_gets - (n_gets_fail + n_gets_ok)); // FIXME: It always prints 1.
  n_gets_ok++;
  get_ac->nrefs--;
  GNUNET_DHT_get_stop (ac->dht_get);
  ac->dht_get = NULL;
  if (ac->delay_task != NULL)
    GNUNET_SCHEDULER_cancel (ac->delay_task);
  ac->delay_task = NULL;
  GNUNET_assert (NULL != ctx->op);
  GNUNET_TESTBED_operation_done (ctx->op);
  ctx->op = NULL;

  total_put_path_length = total_put_path_length + (double) put_path_length;
  total_get_path_length = total_get_path_length + (double) get_path_length;
  DEBUG ("total_put_path_length = %u,put_path \n",
         total_put_path_length);
  /* Summarize if profiling is complete */
  if (n_active == n_gets_fail + n_gets_ok)
  {
    average_put_path_length = (double) total_put_path_length
                              / (double) n_active;
    average_get_path_length = (double) total_get_path_length
                              / (double ) n_gets_ok;
    summarize ();
  }
}


static void
delayed_get (void *cls)
{
  struct ActiveContext *ac = cls;
  struct ActiveContext *get_ac;
  unsigned int r;

  ac->delay_task = NULL;
  get_ac = NULL;
  while (1)
  {
    r = GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
                                  n_active);
    get_ac = &a_ac[r];
    if (NULL != get_ac->hash)
      break;
  }
  get_ac->nrefs++;
  ac->get_ac = get_ac;
  r = GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
                                num_puts_per_peer);
  DEBUG ("GET_REQUEST_START key %s \n",
         GNUNET_h2s (&get_ac->hash[r]));
  ac->dht_get = GNUNET_DHT_get_start (ac->dht,
                                      GNUNET_BLOCK_TYPE_TEST,
                                      &get_ac->hash[r],
                                      1, /* replication level */
                                      GNUNET_DHT_RO_NONE,
                                      NULL,
                                      0, /* extended query and size */
                                      &get_iter,
                                      ac); /* GET iterator and closure */
  n_gets++;

  /* schedule the timeout task for GET */
  ac->delay_task = GNUNET_SCHEDULER_add_delayed (timeout,
                                                 &cancel_get,
                                                 ac);
}


static void
delayed_put (void *cls);


static void
put_cont (void *cls)
{
  struct ActiveContext *ac = cls;

  ac->dht_put = NULL;
  n_puts_ok++;
  ac->delay_task = GNUNET_SCHEDULER_add_now (&delayed_put,
                                             ac);
}


static void
delayed_put (void *cls)
{
  struct ActiveContext *ac = cls;
  char block[65536];
  size_t block_size;

  ac->delay_task = NULL;
  if (0 == ac->put_count)
  {
    struct Context *ctx = ac->ctx;
    struct GNUNET_TESTBED_Operation *op;

    GNUNET_assert (NULL != ctx);
    op = ctx->op;
    ctx->op = NULL;
    GNUNET_TESTBED_operation_done (op);
    return;
  }


  /* Generate and DHT PUT some random data */
  block_size = 16; /* minimum */
  /* make random payload, reserve 512 - 16 bytes for DHT headers */
  block_size += GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
                                          GNUNET_CONSTANTS_MAX_ENCRYPTED_MESSAGE_SIZE
                                          - 512);
  GNUNET_CRYPTO_random_block (GNUNET_CRYPTO_QUALITY_WEAK,
                              block,
                              block_size);
  ac->put_count--;
  GNUNET_CRYPTO_hash (block,
                      block_size,
                      &ac->hash[ac->put_count]);
  DEBUG ("PUT_REQUEST_START key %s\n",
         GNUNET_h2s (&ac->hash[ac->put_count]));
  ac->dht_put = GNUNET_DHT_put (ac->dht,
                                &ac->hash[ac->put_count],
                                replication,
                                GNUNET_DHT_RO_RECORD_ROUTE,
                                GNUNET_BLOCK_TYPE_TEST,
                                block_size,
                                block,
                                GNUNET_TIME_UNIT_FOREVER_ABS, /* expiration time */
                                &put_cont,
                                ac);                /* continuation and its closure */
  n_puts++;
}


static void
dht_connected (void *cls,
               struct GNUNET_TESTBED_Operation *op,
               void *ca_result,
               const char *emsg)
{
  struct ActiveContext *ac = cls;
  struct Context *ctx = ac->ctx;

  GNUNET_assert (NULL != ctx);  // FIXME: Fails
  GNUNET_assert (NULL != ctx->op);
  GNUNET_assert (ctx->op == op);
  ac->dht = (struct GNUNET_DHT_Handle *) ca_result;
  if (NULL != emsg)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
                "Connection to DHT service failed: %s\n",
                emsg);
    GNUNET_TESTBED_operation_done (ctx->op);  /* Calls dht_disconnect() */
    ctx->op = NULL;
    return;
  }
  switch (mode)
  {
  case MODE_PUT:
    {
      struct GNUNET_TIME_Relative peer_delay_put;

      peer_delay_put.rel_value_us =
        GNUNET_CRYPTO_random_u64 (GNUNET_CRYPTO_QUALITY_WEAK,
                                  delay_put.rel_value_us);
      ac->put_count = num_puts_per_peer;
      ac->hash = calloc (ac->put_count,
                         sizeof(struct GNUNET_HashCode));
      if (NULL == ac->hash)
      {
        GNUNET_log_strerror (GNUNET_ERROR_TYPE_ERROR,
                             "calloc");
        GNUNET_SCHEDULER_shutdown ();
        return;
      }
      ac->delay_task = GNUNET_SCHEDULER_add_delayed (peer_delay_put,
                                                     &delayed_put,
                                                     ac);
      break;
    }

  case MODE_GET:
    {
      struct GNUNET_TIME_Relative peer_delay_get;

      peer_delay_get.rel_value_us =
        delay_get.rel_value_us
        + GNUNET_CRYPTO_random_u64 (GNUNET_CRYPTO_QUALITY_WEAK,
                                    delay_get.rel_value_us);
      ac->delay_task = GNUNET_SCHEDULER_add_delayed (peer_delay_get,
                                                     &delayed_get,
                                                     ac);
      break;
    }
  }
}


static void *
dht_connect (void *cls,
             const struct GNUNET_CONFIGURATION_Handle *cfg)
{
  n_dht++;
  return GNUNET_DHT_connect (cfg,
                             10);
}


static void
dht_disconnect (void *cls,
                void *op_result)
{
  struct ActiveContext *ac = cls;

  GNUNET_assert (NULL != ac->dht);
  GNUNET_assert (ac->dht == op_result);
  GNUNET_DHT_disconnect (ac->dht);
  ac->dht = NULL;
  n_dht--;
  if (0 != n_dht)
    return;
  if (GNUNET_YES == in_shutdown)
    return;
  switch (mode)
  {
  case MODE_PUT:
    if (n_puts_ok != ((unsigned long long) n_active) * num_puts_per_peer)
      return;
    /* Start GETs if all PUTs have been made */
    mode = MODE_GET;
    start_profiling ();
    return;

  case MODE_GET:
    if ((n_gets_ok + n_gets_fail) != n_active)
      return;
    break;
  }
}


static void
start_profiling ()
{
  struct Context *ctx;

  DEBUG ("GNUNET_TESTBED_service_connect\n");
  GNUNET_break (GNUNET_YES != in_shutdown);
  for (unsigned int i = 0; i < n_active; i++)
  {
    struct ActiveContext *ac = &a_ac[i];
    GNUNET_assert (NULL != (ctx = ac->ctx));
    GNUNET_assert (NULL == ctx->op);
    ctx->op = GNUNET_TESTBED_service_connect (ctx,
                                              ctx->peer,
                                              "dht",
                                              &dht_connected, ac,
                                              &dht_connect,
                                              &dht_disconnect,
                                              ac);
  }
}


static void
service_started (void *cls,
                 struct GNUNET_TESTBED_Operation *op,
                 const char *emsg)
{
  struct Context *ctx = cls;

  GNUNET_assert (NULL != ctx);
  GNUNET_assert (NULL != ctx->op);
  GNUNET_TESTBED_operation_done (ctx->op);
  ctx->op = NULL;
  peers_started++;
  DEBUG ("Peers Started = %d; num_peers = %d \n",
         peers_started,
         num_peers);
  if (peers_started == num_peers)
    start_profiling ();
}


static void
test_run (void *cls,
          struct GNUNET_TESTBED_RunHandle *h,
          unsigned int num_peers,
          struct GNUNET_TESTBED_Peer **peers,
          unsigned int links_succeeded,
          unsigned int links_failed)
{
  unsigned int ac_cnt;

  testbed_handles = peers;
  if (NULL == peers)
  {
    /* exit */
    GNUNET_assert (0);
  }
  MESSAGE ("%u peers started, %u/%u links up\n",
           num_peers,
           links_succeeded,
           links_succeeded + links_failed);
  a_ctx = GNUNET_new_array (num_peers,
                            struct Context);
  /* select the peers which actively participate in profiling */
  n_active = num_peers * put_probability / 100;
  if (0 == n_active)
  {
    GNUNET_SCHEDULER_shutdown ();
    GNUNET_free (a_ctx);
    a_ctx = NULL;
    return;
  }

  a_ac = GNUNET_new_array (n_active,
                           struct ActiveContext);
  ac_cnt = 0;
  for (unsigned int cnt = 0; cnt < num_peers && ac_cnt < n_active; cnt++)
  {
    if (GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
                                  100) >= put_probability)
      continue;

    a_ctx[cnt].ac = &a_ac[ac_cnt];
    a_ac[ac_cnt].ctx = &a_ctx[cnt];
    ac_cnt++;
  }
  n_active = ac_cnt;

  /* start DHT service on all peers */
  for (unsigned int cnt = 0; cnt < num_peers; cnt++)
  {
    a_ctx[cnt].peer = peers[cnt];
    a_ctx[cnt].op = GNUNET_TESTBED_peer_manage_service (&a_ctx[cnt],
                                                        peers[cnt],
                                                        "dht",
                                                        &service_started,
                                                        &a_ctx[cnt],
                                                        1);
  }
}


static void
run (void *cls,
     char *const *args,
     const char *cfgfile,
     const struct GNUNET_CONFIGURATION_Handle *config)
{
  uint64_t event_mask;

  if (0 == num_peers)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
                _ ("Exiting as the number of peers is %u\n"),
                num_peers);
    return;
  }
  cfg = config;
  event_mask = 0;
  GNUNET_TESTBED_run (hosts_file,
                      cfg,
                      num_peers,
                      event_mask,
                      NULL,
                      NULL,
                      &test_run,
                      NULL);
  GNUNET_SCHEDULER_add_shutdown (&do_shutdown,
                                 NULL);
}


int
main (int argc,
      char *const *argv)
{
  int rc;
  struct GNUNET_GETOPT_CommandLineOption options[] = {
    GNUNET_GETOPT_option_uint ('n',
                               "peers",
                               "COUNT",
                               gettext_noop ("number of peers to start"),
                               &num_peers),
    GNUNET_GETOPT_option_uint ('p',
                               "peer-put-count",
                               "COUNT",
                               gettext_noop (
                                 "number of PUTs to perform per peer"),
                               &num_puts_per_peer),
    GNUNET_GETOPT_option_string ('H',
                                 "hosts",
                                 "FILENAME",
                                 gettext_noop (
                                   "name of the file with the login information for the testbed"),
                                 &hosts_file),
    GNUNET_GETOPT_option_relative_time ('D',
                                        "delay",
                                        "DELAY",
                                        gettext_noop (
                                          "delay between rounds for collecting statistics (default: 30 sec)"),
                                        &delay_stats),
    GNUNET_GETOPT_option_relative_time ('P',
                                        "PUT-delay",
                                        "DELAY",
                                        gettext_noop (
                                          "delay to start doing PUTs (default: 1 sec)"),
                                        &delay_put),
    GNUNET_GETOPT_option_relative_time ('G',
                                        "GET-delay",
                                        "DELAY",
                                        gettext_noop (
                                          "delay to start doing GETs (default: 5 min)"),
                                        &delay_get),
    GNUNET_GETOPT_option_uint ('r',
                               "replication",
                               "DEGREE",
                               gettext_noop ("replication degree for DHT PUTs"),
                               &replication),
    GNUNET_GETOPT_option_uint ('R',
                               "random-chance",
                               "PROBABILITY",
                               gettext_noop (
                                 "chance that a peer is selected at random for PUTs"),
                               &put_probability),
    GNUNET_GETOPT_option_relative_time ('t',
                                        "timeout",
                                        "TIMEOUT",
                                        gettext_noop (
                                          "timeout for DHT PUT and GET requests (default: 1 min)"),
                                        &timeout),
    GNUNET_GETOPT_OPTION_END
  };

  if (GNUNET_OK !=
      GNUNET_STRINGS_get_utf8_args (argc, argv,
                                    &argc, &argv))
    return 2;
  /* set default delays */
  delay_stats = GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 10);
  delay_put = GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 10);
  delay_get = GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 10);
  timeout = GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 10);
  replication = 1;      /* default replication */
  rc = 0;
  if (GNUNET_OK !=
      GNUNET_PROGRAM_run (argc,
                          argv,
                          "gnunet-dht-profiler",
                          gettext_noop (
                            "Measure quality and performance of the DHT service."),
                          options,
                          &run,
                          NULL))
    rc = 1;
  return rc;
}