/**
 * Function setting up file descriptors and scheduling task to run
 *
 * @param  plugin plugin as closure
 * @param now schedule task in 1ms, regardless of what curl may say
 * @return GNUNET_SYSERR for hard failure, GNUNET_OK for ok
 */
static int
client_schedule (struct Plugin *plugin, int now)
{
  fd_set rs;
  fd_set ws;
  fd_set es;
  int max;
  struct GNUNET_NETWORK_FDSet *grs;
  struct GNUNET_NETWORK_FDSet *gws;
  long to;
  CURLMcode mret;
  struct GNUNET_TIME_Relative timeout;

  /* Cancel previous scheduled task */
  if (plugin->client_perform_task != GNUNET_SCHEDULER_NO_TASK)
  {
    GNUNET_SCHEDULER_cancel (plugin->client_perform_task);
    plugin->client_perform_task = GNUNET_SCHEDULER_NO_TASK;
  }
  max = -1;
  FD_ZERO (&rs);
  FD_ZERO (&ws);
  FD_ZERO (&es);
  mret = curl_multi_fdset (plugin->client_mh, &rs, &ws, &es, &max);
  if (mret != CURLM_OK)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_ERROR, _("%s failed at %s:%d: `%s'\n"),
                "curl_multi_fdset", __FILE__, __LINE__,
                curl_multi_strerror (mret));
    return GNUNET_SYSERR;
  }
  mret = curl_multi_timeout (plugin->client_mh, &to);
  if (to == -1)
    timeout = GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 1);
  else
    timeout = GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MILLISECONDS, to);
  if (now == GNUNET_YES)
    timeout = GNUNET_TIME_UNIT_MILLISECONDS;

  if (mret != CURLM_OK)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_ERROR, _("%s failed at %s:%d: `%s'\n"),
                "curl_multi_timeout", __FILE__, __LINE__,
                curl_multi_strerror (mret));
    return GNUNET_SYSERR;
  }

  grs = GNUNET_NETWORK_fdset_create ();
  gws = GNUNET_NETWORK_fdset_create ();
  GNUNET_NETWORK_fdset_copy_native (grs, &rs, max + 1);
  GNUNET_NETWORK_fdset_copy_native (gws, &ws, max + 1);

  plugin->client_perform_task =
      GNUNET_SCHEDULER_add_select (GNUNET_SCHEDULER_PRIORITY_DEFAULT,
                                   timeout, grs, gws,
                                   &client_run, plugin);
  GNUNET_NETWORK_fdset_destroy (gws);
  GNUNET_NETWORK_fdset_destroy (grs);
  return GNUNET_OK;
}

/**
 * Signature of the main function of a task.
 *
 * @param cls closure
 */
static void
test_task (void *cls)
{
  struct GNUNET_TIME_Absolute now;

  now = GNUNET_TIME_absolute_get ();
  if (now.abs_value_us > target.abs_value_us)
    cumDelta += (now.abs_value_us - target.abs_value_us);
  else
    cumDelta += (target.abs_value_us - now.abs_value_us);
  target =
      GNUNET_TIME_relative_to_absolute (GNUNET_TIME_relative_multiply
                                        (GNUNET_TIME_UNIT_MICROSECONDS, i));
  FPRINTF (stderr, "%s",  ".");
  if (i > MAXV)
  {
    FPRINTF (stderr, "%s",  "\n");
    return;
  }
  GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_relative_multiply
                                (GNUNET_TIME_UNIT_MICROSECONDS, i),
				&test_task,
                                NULL);
  i += INCR;
}

static void
peers_started_callback (void *cls, const struct GNUNET_PeerIdentity *id,
                        const struct GNUNET_CONFIGURATION_Handle *cfg,
                        struct GNUNET_TESTING_Daemon *d, const char *emsg)
{
  if (emsg != NULL)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Failed to start daemon with error: `%s'\n", emsg);
    return;
  }
  GNUNET_assert (id != NULL);
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Started daemon %llu out of %llu\n",
              (num_peers - peers_left) + 1, num_peers);
#if PROGRESS_BARS
  if ((num_peers - peers_left) % modnum == 0)
  {
    if (num_peers - peers_left == 0)
      FPRINTF (stdout, "%s",  "0%%");
    else
      FPRINTF (stdout, "%d%%",
               (int) (((float) (num_peers - peers_left) / num_peers) * 100));

  }
  else if ((num_peers - peers_left) % dotnum == 0)
  {
    FPRINTF (stdout, "%s",  ".");
  }
  fflush (stdout);
#endif
  peers_left--;
  if (peers_left == 0)
  {
#if PROGRESS_BARS
    FPRINTF (stdout, "%s",  "100%%]\n");
#endif
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "All %d daemons started, now connecting peers!\n", num_peers);
    GNUNET_SCHEDULER_cancel (die_task);
    /* Set up task in case topology creation doesn't finish
     * within a reasonable amount of time */
    die_task =
        GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_relative_multiply
                                      (GNUNET_TIME_UNIT_MINUTES, 8), &end_badly,
                                      "from peers_started_callback");
#if DELAY_FOR_LOGGING
    FPRINTF (stdout, "%s",  "Connecting topology in 10 seconds\n");
    gather_log_data ();
    GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_relative_multiply
                                  (GNUNET_TIME_UNIT_SECONDS, 10),
                                  &connect_topology, NULL);
#else
    connect_topology ();
#endif
    ok = 0;
  }
}

static void
send_test_messages (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
{
  struct TestMessageContext *pos = cls;

  if ((pos == test_messages) && (settle_time.rel_value > 0))
  {
    topology_connections = 0;
    GNUNET_TESTING_get_topology (pg, &topology_cb, NULL);
  }
  if (((tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN) != 0) || (cls == NULL))
    return;

  if (die_task == GNUNET_SCHEDULER_NO_TASK)
  {
    die_task =
        GNUNET_SCHEDULER_add_delayed (TEST_TIMEOUT, &end_badly,
                                      "from send test messages (timeout)");
  }

  if (total_server_connections >= MAX_OUTSTANDING_CONNECTIONS)
  {
    GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_relative_multiply
                                  (GNUNET_TIME_UNIT_SECONDS, 1),
                                  &send_test_messages, pos);
    return;                     /* Otherwise we'll double schedule messages here! */
  }

  /*
   * Connect to the sending peer
   */
  pos->peer1handle =
      GNUNET_CORE_connect (pos->peer1->cfg, pos, &init_notify_peer1,
                           &connect_notify_peers, NULL, NULL, GNUNET_NO, NULL,
                           GNUNET_NO, no_handlers);

  GNUNET_assert (pos->peer1handle != NULL);

  if (total_server_connections < MAX_OUTSTANDING_CONNECTIONS)
  {
    GNUNET_SCHEDULER_add_now (&send_test_messages, pos->next);
  }
  else
  {
    GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_relative_multiply
                                  (GNUNET_TIME_UNIT_SECONDS, 1),
                                  &send_test_messages, pos->next);
  }
}

static void
run (void *cls,
     char *const *args,
     const char *cfgfile,
     const struct GNUNET_CONFIGURATION_Handle *cfg)
{
  struct GNUNET_MQ_MessageHandler handlers[] = {
    GNUNET_MQ_hd_fixed_size (test,
                             MTYPE,
                             struct GNUNET_MessageHeader,
                             NULL),
    GNUNET_MQ_handler_end ()
  };

  GNUNET_assert (ok == 1);
  OKPP;
  setup_peer (&p1,
	      "test_core_api_peer1.conf");
  setup_peer (&p2,
	      "test_core_api_peer2.conf");
  err_task =
      GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_relative_multiply
                                    (GNUNET_TIME_UNIT_SECONDS, 300),
                                    &terminate_task_error, NULL);
  p1.ch =
      GNUNET_CORE_connect (p1.cfg,
			   &p1,
                           &init_notify,
                           &connect_notify,
                           &disconnect_notify,
                           handlers);
}

static void
create_topology ()
{
  peers_left = num_peers;       /* Reset counter */
  if (GNUNET_TESTING_create_topology
      (pg, topology, blacklist_topology, blacklist_transports) != GNUNET_SYSERR)
  {
#if PROGRESS_BARS
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Topology set up, now starting peers!\n");
    FPRINTF (stdout, "%s",  "Daemon start progress [");
#endif
    GNUNET_TESTING_daemons_continue_startup (pg);
  }
  else
  {
    GNUNET_SCHEDULER_cancel (die_task);
    die_task =
        GNUNET_SCHEDULER_add_now (&end_badly,
                                  "from create topology (bad return)");
  }
  GNUNET_SCHEDULER_cancel (die_task);
  die_task =
      GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_relative_multiply
                                    (GNUNET_TIME_UNIT_SECONDS,
                                     SECONDS_PER_PEER_START * num_peers),
                                    &end_badly,
                                    "from continue startup (timeout)");
}

static void
test_master (void *cls,
	     struct GNUNET_TESTBED_RunHandle *h,
	     unsigned int num_peers,
	     struct GNUNET_TESTBED_Peer **peers,
	     unsigned int links_succeeded,
	     unsigned int links_failed)
{
  basic_mqtt_op_publish = GNUNET_TESTBED_service_connect(NULL, /* Closure for operation */
							 peers[0], /* The peer whose service to connect to */
							 "gnunet-service-mqtt", /* The name of the service */
							 service_connect_comp_publish, /* callback to call after a handle to service is opened */
							 NULL, /* closure for the above callback */
							 gmqtt_ca_publish, /* callback to call with peer's configuration; this should open the needed service connection */
							 gmqtt_da_publish, /* callback to be called when closing the opened service connection */
							 NULL); /* closure for the above two callbacks */
  basic_mqtt_op_subscribe = GNUNET_TESTBED_service_connect(NULL, /* Closure for operation */
							   peers[1], /* The peer whose service to connect to */
							   "gnunet-service-mqtt", /* The name of the service */
							   service_connect_comp_subscribe, /* callback to call after a handle to service is opened */
							   NULL, /* closure for the above callback */
							   gmqtt_ca_subscribe, /* callback to call with peer's configuration; this should open the needed service connection */
							   gmqtt_da_subscribe, /* callback to be called when closing the opened service connection */
							   NULL); /* closure for the above two callbacks */			
  
  shutdown_tid = GNUNET_SCHEDULER_add_delayed(GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 10), 
					      &shutdown_task, NULL);
}

/**
 * Main function that will be run by the scheduler.
 *
 * @param cls closure
 * @param args remaining command-line arguments
 * @param cfgfile name of the configuration file used (for saving, can be NULL!)
 * @param cfg configuration
 */
static void
run (void *cls, char *const *args, const char *cfgfile,
     const struct GNUNET_CONFIGURATION_Handle *cfg)
{
  struct GNUNET_TIME_Relative timeout;

  if ((NULL == topic))
  {
    FPRINTF (stderr, "%s",
             _("Must provide TOPIC for MQTT SUBSCRIBE!\n"));
    ret = 1;
    return;
  }

  mqtt_handle = GNUNET_MQTT_connect (cfg);

  if (NULL == mqtt_handle)
  {
    FPRINTF (stderr, "%s",  _("Failed to connect to MQTT service!\n"));
    ret = 1;
    return;
  }

  timeout = GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS,
                                           request_timeout);

  GNUNET_MQTT_subscribe (mqtt_handle, strlen(topic) + 1, topic, timeout,
                         subscribe_continuation, NULL,
                         subscribe_result_callback, NULL);

  ret = 0;
}

/**
 * Initialize framework and start test
 *
 * @param cls closure
 * @param cfg configuration of the peer that was started
 * @param peer identity of the peer that was created
 */
static void
run (void *cls,
     const struct GNUNET_CONFIGURATION_Handle *cfg,
     struct GNUNET_TESTING_Peer *peer)
{
    struct GNUNET_PeerIdentity self;

    GNUNET_TESTING_peer_get_identity (peer, &self);
    config = cfg;
    peer2_listen_socket =
        GNUNET_STREAM_listen (config,
                              10, /* App port */
                              &stream_listen_cb,
                              NULL,
                              GNUNET_STREAM_OPTION_SIGNAL_LISTEN_SUCCESS,
                              &stream_connect,
                              GNUNET_STREAM_OPTION_END);
    GNUNET_assert (NULL != peer2_listen_socket);
    peer1.self = self;
    peer2.self = self;
    abort_task =
        GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_relative_multiply
                                      (GNUNET_TIME_UNIT_SECONDS, 60), &do_abort,
                                      NULL);
}

/**
 * The write completion function; called upon writing some data to stream or
 * upon error
 *
 * @param cls the closure from GNUNET_STREAM_write/read
 * @param status the status of the stream at the time this function is called
 * @param size the number of bytes read or written
 */
static void
write_completion (void *cls,
                  enum GNUNET_STREAM_Status status,
                  size_t size)
{
    struct PeerData *peer;

    peer = (struct PeerData *) cls;
    GNUNET_assert (GNUNET_STREAM_OK == status);
    GNUNET_assert (size <= DATA_SIZE);
    peer->bytes_wrote += size;

    if (peer->bytes_wrote < DATA_SIZE) /* Have more data to send */
    {
        peer->io_write_handle =
            GNUNET_STREAM_write (peer->socket,
                                 ((void *) data) + peer->bytes_wrote,
                                 sizeof (data) - peer->bytes_wrote,
                                 GNUNET_TIME_relative_multiply
                                 (GNUNET_TIME_UNIT_SECONDS, 5),
                                 &write_completion,
                                 cls);
        GNUNET_assert (NULL != peer->io_write_handle);
    }
    else
    {
        LOG (GNUNET_ERROR_TYPE_DEBUG, "Writing successfully finished\n");
        result = GNUNET_OK;
        GNUNET_SCHEDULER_add_now (&do_shutdown, NULL);
    }
}

/**
 * Main run function.
 *
 * @param cls NULL
 * @param args arguments passed to GNUNET_PROGRAM_run
 * @param cfgfile the path to configuration file
 * @param cfg the configuration file handle
 */
static void
run (void *cls, char *const *args, const char *cfgfile,
     const struct GNUNET_CONFIGURATION_Handle *config)
{
  cfg = GNUNET_CONFIGURATION_dup (config);
  host = GNUNET_TESTBED_host_create (NULL, NULL, cfg, 0);
  FAIL_TEST (NULL != host);
  if (NULL ==
      (hc_handle =
       GNUNET_TESTBED_is_host_habitable (host, config, &host_habitable_cb,
                                         NULL)))
  {
    GNUNET_TESTBED_host_destroy (host);
    GNUNET_CONFIGURATION_destroy (cfg);
    cfg = NULL;
    host = NULL;
    (void) PRINTF ("%s",
                   "Unable to run the test as this system is not configured "
                   "to use password less SSH logins to localhost.\n"
                   "Marking test as successful\n");
    result = SKIP;
    return;
  }
  abort_task =
      GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_relative_multiply
                                    (GNUNET_TIME_UNIT_MINUTES, 5), &do_abort,
                                    NULL);
}

static void
peers_started_callback (void *cls, const struct GNUNET_PeerIdentity *id,
                        const struct GNUNET_CONFIGURATION_Handle *cfg,
                        struct GNUNET_TESTING_Daemon *d, const char *emsg)
{
  if (emsg != NULL)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Failed to start daemon with error: `%s'\n", emsg);
    return;
  }
  GNUNET_assert (id != NULL);
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Started daemon %llu out of %llu\n",
              (num_peers - peers_left) + 1, num_peers);
  peers_left--;
  if (peers_left == 0)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "All %d daemons started, now testing churn!\n", num_peers);
    GNUNET_SCHEDULER_cancel (die_task);
    /* Set up task in case topology creation doesn't finish
     * within a reasonable amount of time */
    die_task =
        GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_relative_multiply
                                      (GNUNET_TIME_UNIT_MINUTES, 5), &end_badly,
                                      "from peers_started_callback");
    churn_peers_off ();
    ok = 0;
  }
}

static void
run_task (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
{
  char *fn;
  const struct GNUNET_DISK_FileHandle *stdout_read_handle;
  const struct GNUNET_DISK_FileHandle *wh;

#if !WINDOWS
  GNUNET_asprintf (&fn, "cat");
#else
  GNUNET_asprintf (&fn, "w32cat");
#endif

  hello_pipe_stdin = GNUNET_DISK_pipe (GNUNET_YES, GNUNET_YES, GNUNET_YES, GNUNET_NO);
  hello_pipe_stdout = GNUNET_DISK_pipe (GNUNET_YES, GNUNET_YES, GNUNET_NO, GNUNET_YES);

  if ((hello_pipe_stdout == NULL) || (hello_pipe_stdin == NULL))
  {
    GNUNET_break (0);
    ok = 1;
    GNUNET_free (fn);
    return;
  }

  proc =
      GNUNET_OS_start_process (GNUNET_NO, GNUNET_OS_INHERIT_STD_ERR, hello_pipe_stdin, hello_pipe_stdout, fn,
                               "test_gnunet_echo_hello", "-", NULL);
  GNUNET_free (fn);

  /* Close the write end of the read pipe */
  GNUNET_DISK_pipe_close_end (hello_pipe_stdout, GNUNET_DISK_PIPE_END_WRITE);
  /* Close the read end of the write pipe */
  GNUNET_DISK_pipe_close_end (hello_pipe_stdin, GNUNET_DISK_PIPE_END_READ);

  wh = GNUNET_DISK_pipe_handle (hello_pipe_stdin, GNUNET_DISK_PIPE_END_WRITE);

  /* Write the test_phrase to the cat process */
  if (GNUNET_DISK_file_write (wh, test_phrase, strlen (test_phrase) + 1) !=
      strlen (test_phrase) + 1)
  {
    GNUNET_break (0);
    ok = 1;
    return;
  }

  /* Close the write end to end the cycle! */
  GNUNET_DISK_pipe_close_end (hello_pipe_stdin, GNUNET_DISK_PIPE_END_WRITE);

  stdout_read_handle =
      GNUNET_DISK_pipe_handle (hello_pipe_stdout, GNUNET_DISK_PIPE_END_READ);

  die_task =
      GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_relative_multiply
                                    (GNUNET_TIME_UNIT_MINUTES, 1), &end_task,
                                    NULL);

  GNUNET_SCHEDULER_add_read_file (GNUNET_TIME_UNIT_FOREVER_REL,
                                  stdout_read_handle, &read_call,
                                  (void *) stdout_read_handle);
}

static void
mhd_main ()
{
  struct GNUNET_NETWORK_FDSet nrs;
  struct GNUNET_NETWORK_FDSet nws;
  fd_set rs;
  fd_set ws;
  fd_set es;
  int max_fd;
  unsigned MHD_LONG_LONG timeout;
  struct GNUNET_TIME_Relative delay;

  GNUNET_assert (NULL == mhd_task_id);
  FD_ZERO (&rs);
  FD_ZERO (&ws);
  FD_ZERO (&es);
  max_fd = -1;
  GNUNET_assert (MHD_YES == MHD_get_fdset (mhd, &rs, &ws, &es, &max_fd));
  if (MHD_YES == MHD_get_timeout (mhd, &timeout))
    delay =
        GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MILLISECONDS,
                                       (unsigned int) timeout);
  else
    delay = GNUNET_TIME_UNIT_FOREVER_REL;
  GNUNET_NETWORK_fdset_copy_native (&nrs, &rs, max_fd + 1);
  GNUNET_NETWORK_fdset_copy_native (&nws, &ws, max_fd + 1);
  mhd_task_id =
      GNUNET_SCHEDULER_add_select (GNUNET_SCHEDULER_PRIORITY_DEFAULT, delay,
                                   &nrs, &nws, &mhd_task, NULL);
}

static void
run (void *cls,
     const struct GNUNET_CONFIGURATION_Handle *cfg,
     struct GNUNET_TESTING_Peer *peer)
{
  struct GNUNET_HashCode hash;
  char *data;
  size_t data_size = 42;

  GNUNET_assert (ok == 1);
  OKPP;
  die_task =
      GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_relative_multiply
                                    (GNUNET_TIME_UNIT_MINUTES, 1), &end_badly,
                                    NULL);


  memset (&hash, 42, sizeof (struct GNUNET_HashCode));
  data = GNUNET_malloc (data_size);
  memset (data, 43, data_size);
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Called test_put!\n");
  dht_handle = GNUNET_DHT_connect (cfg, 100);
  GNUNET_assert (dht_handle != NULL);
  GNUNET_DHT_put (dht_handle, &hash, 1, GNUNET_DHT_RO_NONE,
                  GNUNET_BLOCK_TYPE_TEST, data_size, data,
                  GNUNET_TIME_relative_to_absolute (TOTAL_TIMEOUT),
                  &test_get, NULL);
  GNUNET_free (data);
}

int blacklist_cb (void *cls,
                 const struct
                 GNUNET_PeerIdentity * pid)
{
  struct PeerContext * p = cls;
  int res = GNUNET_SYSERR;
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Peer  %u : Blacklist request for peer `%s'\n", p->no, GNUNET_i2s (pid));

  if (p == p1)
  {
    blacklist_request_p1 = GNUNET_YES;
    res = GNUNET_OK;
  }
  else if (p == p2)
  {
    blacklist_request_p2 = GNUNET_YES;
    res = GNUNET_SYSERR;
  }

  if (((blacklist_request_p2 == GNUNET_YES) && (blacklist_request_p1 == GNUNET_YES)) && (shutdown_task == GNUNET_SCHEDULER_NO_TASK))
  {
    shutdown_task = GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_relative_multiply(GNUNET_TIME_UNIT_SECONDS, 3), &end, NULL);
  }

  return res;
}

/**
 * Task for calling STREAM_write with a chunk of random data
 *
 * @param cls the peer data entity
 * @param tc the task context
 */
static void
stream_write_task (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
{
  struct PeerData *pdata = cls;
  size_t write_amount;

  if (GNUNET_SCHEDULER_NO_TASK != abort_task)
  {
    GNUNET_SCHEDULER_cancel (abort_task);
    abort_task = 
      GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_relative_multiply
                                    (GNUNET_TIME_UNIT_SECONDS, 300), &do_abort,
                                    NULL);
  }
  write_task = GNUNET_SCHEDULER_NO_TASK;
  prof_start_time = GNUNET_TIME_absolute_get ();
  pdata->bytes_wrote = 0;
  pdata->packets_wrote = 0;
  write_amount = MAX_PACKETS * payload_size[payload_size_index];
  if (write_amount > DATA_SIZE)
    write_amount = DATA_SIZE;
  reset_read = GNUNET_YES;
  pdata->io_write_handle = GNUNET_STREAM_write (pdata->socket, data,
                                                write_amount,
						GNUNET_TIME_UNIT_FOREVER_REL,
						&write_completion, pdata);
  GNUNET_assert (NULL != pdata->io_write_handle);
}

/**
 * Signature of a main function for a testcase.
 *
 * @param cls closure
 * @param h the run handle
 * @param num_peers number of peers in 'peers'
 * @param peers_ handle to peers run in the testbed
 * @param links_succeeded the number of overlay link connection attempts that
 *          succeeded
 * @param links_failed the number of overlay link connection attempts that
 *          failed
 */
static void
test_master (void *cls,
             struct GNUNET_TESTBED_RunHandle *h,
             unsigned int num_peers,
             struct GNUNET_TESTBED_Peer **peers_,
             unsigned int links_succeeded,
             unsigned int links_failed)
{
  struct GNUNET_TESTBED_Controller *c;

  GNUNET_assert (NULL == cls);
  if (NULL == peers_)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "Failing test due to timeout\n");
    return;
  }
  GNUNET_assert (NUM_PEERS == num_peers);
  c = GNUNET_TESTBED_run_get_controller_handle (h);
  barrier = GNUNET_TESTBED_barrier_init (c, TEST_BARRIER_NAME, 100,
                                         &barrier_cb, NULL);
  shutdown_task =
      GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_relative_multiply
                                    (GNUNET_TIME_UNIT_SECONDS,
                                     10 * (NUM_PEERS + 1)),
                                    &do_shutdown, NULL);
}

static void
run_accept (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
{
  void *addr;
  size_t alen;
  struct sockaddr_in *v4;
  struct sockaddr_in expect;

  asock = GNUNET_CONNECTION_create_from_accept (NULL, NULL, ls);
  GNUNET_assert (asock != NULL);
  GNUNET_assert (GNUNET_YES == GNUNET_CONNECTION_check (asock));
  GNUNET_assert (GNUNET_OK ==
                 GNUNET_CONNECTION_get_address (asock, &addr, &alen));
  GNUNET_assert (alen == sizeof (struct sockaddr_in));
  v4 = addr;
  memset (&expect, 0, sizeof (expect));
#if HAVE_SOCKADDR_IN_SIN_LEN
  expect.sin_len = sizeof (expect);
#endif
  expect.sin_family = AF_INET;
  expect.sin_port = v4->sin_port;
  expect.sin_addr.s_addr = htonl (INADDR_LOOPBACK);
  GNUNET_assert (0 == memcmp (&expect, v4, alen));
  GNUNET_free (addr);
  GNUNET_CONNECTION_destroy (lsock);
  GNUNET_CONNECTION_receive (asock, 1024,
                             GNUNET_TIME_relative_multiply
                             (GNUNET_TIME_UNIT_SECONDS, 5), &receive_check,
                             cls);
}

/**
 * Main run function.
 *
 * @param cls NULL
 * @param args arguments passed to GNUNET_PROGRAM_run
 * @param cfgfile the path to configuration file
 * @param cfg the configuration file handle
 */
static void
run (void *cls, char *const *args, const char *cfgfile,
     const struct GNUNET_CONFIGURATION_Handle *config)
{
  host = GNUNET_TESTBED_host_create (NULL, NULL, 0);
  GNUNET_assert (NULL != host);
  if (GNUNET_YES != GNUNET_TESTBED_is_host_habitable (host, config))
  {
    GNUNET_TESTBED_host_destroy (host);
    host = NULL;
    (void) PRINTF ("%s",
                   "Unable to run the test as this system is not configured "
                   "to use password less SSH logins to localhost.\n"
                   "Marking test as successful\n");
    result = SUCCESS;
    return;
  }
  cfg = GNUNET_CONFIGURATION_dup (config);
  cp = GNUNET_TESTBED_controller_start ("127.0.0.1", host, cfg, status_cb,
                                        NULL);
  abort_task =
      GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_relative_multiply
                                    (GNUNET_TIME_UNIT_MINUTES, 5), &do_abort,
                                    NULL);
}