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_non_null(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;
}