/**
 * Store an item in the datastore.
 *
 * @param h handle to the datacache
 * @param key key to store data under
 * @param size number of bytes in data
 * @param data data to store
 * @param type type of the value
 * @param discard_time when to discard the value in any case
 * @param path_info_len number of entries in 'path_info'
 * @param path_info a path through the network
 * @return GNUNET_OK on success, GNUNET_SYSERR on error, GNUNET_NO if duplicate
 */
int
GNUNET_DATACACHE_put (struct GNUNET_DATACACHE_Handle *h,
                      const struct GNUNET_HashCode * key, size_t size,
                      const char *data, enum GNUNET_BLOCK_Type type,
                      struct GNUNET_TIME_Absolute discard_time,
		      unsigned int path_info_len,
		      const struct GNUNET_PeerIdentity *path_info)
{
  ssize_t used;

  used = h->api->put (h->api->cls, key, 
		      size, data, 
		      type, discard_time,
		      path_info_len, path_info);
  if (-1 == used)
  {
    GNUNET_break (0);
    return GNUNET_SYSERR;
  }
  if (0 == used)
  {
    /* duplicate */
    return GNUNET_NO;
  }
  LOG (GNUNET_ERROR_TYPE_DEBUG, "Stored data under key `%s' in cache\n",
       GNUNET_h2s (key));
  GNUNET_STATISTICS_update (h->stats, gettext_noop ("# bytes stored"), size,
                            GNUNET_NO);
  GNUNET_STATISTICS_update (h->stats, gettext_noop ("# items stored"), 1,
                            GNUNET_NO);
  if (NULL != h->filter)
    GNUNET_CONTAINER_bloomfilter_add (h->filter, key);
  while (h->utilization + used > h->env.quota)
    GNUNET_assert (GNUNET_OK == h->api->del (h->api->cls));
  h->utilization += used;
  return GNUNET_OK;
}

/**
 * Delete an existing identity.
 *
 * @param id identity service to use
 * @param name name of the identity to delete
 * @param cb function to call with the result (will only be called once)
 * @param cb_cls closure for @a cb
 * @return handle to abort the operation
 */
struct GNUNET_IDENTITY_Operation *
GNUNET_IDENTITY_delete (struct GNUNET_IDENTITY_Handle *id,
			const char *name,
			GNUNET_IDENTITY_Continuation cb,
			void *cb_cls)
{
  struct GNUNET_IDENTITY_Operation *op;
  struct GNUNET_IDENTITY_DeleteMessage *gdm;
  size_t slen;

  slen = strlen (name) + 1;
  if (slen >= GNUNET_SERVER_MAX_MESSAGE_SIZE - sizeof (struct GNUNET_IDENTITY_DeleteMessage))
  {
    GNUNET_break (0);
    return NULL;
  }
  op = GNUNET_malloc (sizeof (struct GNUNET_IDENTITY_Operation) +
		      sizeof (struct GNUNET_IDENTITY_DeleteMessage) +
		      slen);
  op->h = id;
  op->cont = cb;
  op->cls = cb_cls;
  gdm = (struct GNUNET_IDENTITY_DeleteMessage *) &op[1];
  gdm->header.type = htons (GNUNET_MESSAGE_TYPE_IDENTITY_DELETE);
  gdm->header.size = htons (sizeof (struct GNUNET_IDENTITY_DeleteMessage) +
			    slen);
  gdm->name_len = htons (slen);
  gdm->reserved = htons (0);
  memcpy (&gdm[1], name, slen);
  op->msg = &gdm->header;
  GNUNET_CONTAINER_DLL_insert_tail (id->op_head,
				    id->op_tail,
				    op);
  if (NULL == id->th)
    transmit_next (id);
  return op;
}

/**
 * Disconnect from the peerinfo service.  Note that all iterators must
 * have completed or have been cancelled by the time this function is
 * called (otherwise, calling this function is a serious error).
 * Furthermore, if 'GNUNET_PEERINFO_add_peer' operations are still
 * pending, they will be cancelled silently on disconnect.
 *
 * @param h handle to disconnect
 */
void
GNUNET_PEERINFO_disconnect (struct GNUNET_PEERINFO_Handle *h)
{
  struct GNUNET_PEERINFO_AddContext *ac;
  struct GNUNET_PEERINFO_IteratorContext *ic;

  while (NULL != (ic = h->ic_head))
  {
    GNUNET_break (GNUNET_YES == ic->in_receive);
    ic->in_receive = GNUNET_NO;
    GNUNET_PEERINFO_iterate_cancel (ic);
  }
  while (NULL != (ac = h->ac_head))
  {
    GNUNET_CONTAINER_DLL_remove (h->ac_head, h->ac_tail, ac);
    if (NULL != ac->cont)
      ac->cont (ac->cont_cls, _("aborted due to explicit disconnect request"));
    GNUNET_free (ac);
  }
  if (NULL != h->th)
  {
    GNUNET_CLIENT_notify_transmit_ready_cancel (h->th);
    h->th = NULL;
  }
  if (NULL != h->client)
  {
    GNUNET_CLIENT_disconnect (h->client);
    h->client = NULL;
  }
  if (GNUNET_SCHEDULER_NO_TASK != h->r_task)
  {
    GNUNET_SCHEDULER_cancel (h->r_task);
    h->r_task = GNUNET_SCHEDULER_NO_TASK;
  }
  GNUNET_free (h);
}

/**
 * Callback which will be called to after a host registration succeeded or failed
 *
 * @param cls the host which has been registered
 * @param emsg the error message; NULL if host registration is successful
 */
static void
registration_cont (void *cls, const char *emsg)
{
  rh = NULL;
  switch (result)
  {
  case MASTER_PEER_START_SUCCESS:
    GNUNET_assert (NULL == emsg);
    GNUNET_assert (NULL != mc);
    result = SLAVE1_REGISTERED;
    slave2 = GNUNET_TESTBED_host_create_with_id (2, "127.0.0.1", NULL, 0);
    GNUNET_assert (NULL != slave2);
    rh = GNUNET_TESTBED_register_host (mc, slave2, &registration_cont, NULL);
    GNUNET_assert (NULL != rh);
    break;
  case SLAVE1_REGISTERED:
    GNUNET_assert (NULL == emsg);
    GNUNET_assert (NULL != mc);
    result = SLAVE2_REGISTERED;
    GNUNET_assert (NULL != cfg);
    op = GNUNET_TESTBED_controller_link (NULL, mc, slave, NULL, cfg, GNUNET_YES);
    GNUNET_assert (NULL != op);
    break;
  case SLAVE2_PEER_DESTROY_SUCCESS:
    GNUNET_assert (NULL == emsg);
    GNUNET_assert (NULL != mc);
    GNUNET_assert (NULL == op);
    result = SLAVE3_REGISTERED;
    op = GNUNET_TESTBED_controller_link (NULL, mc, slave3, NULL, cfg, GNUNET_YES);
    GNUNET_assert (NULL != op);
    break;
  default:
    GNUNET_break (0);
    do_abort_now (NULL);
  }
}

/**
 * Update statistics
 *
 * @param m peermap to update values from
 */
static void
update_stats (struct GNUNET_CONTAINER_MultiPeerMap *m)
{
  GNUNET_assert (NULL != m);
  GNUNET_assert (NULL != GED_stats);

  if (m == nodes_active)
  {
    GNUNET_STATISTICS_set (GED_stats, "# nodes active",
			   GNUNET_CONTAINER_multipeermap_size(m), GNUNET_NO);
  }
  else if (m == nodes_inactive)
  {
    GNUNET_STATISTICS_set (GED_stats, "# nodes inactive",
			   GNUNET_CONTAINER_multipeermap_size(m), GNUNET_NO);
  }
  else if (m == nodes_requested)
  {
    GNUNET_STATISTICS_set (GED_stats, "# nodes requested",
			   GNUNET_CONTAINER_multipeermap_size(m), GNUNET_NO);
  }
  else
    GNUNET_break (0);
}

/**
 * Find a path to a peer that offers a regex service compatible
 * with a given string.
 *
 * @param key The key of the accepting state.
 * @param ctx Context containing info about the string, tunnel, etc.
 */
static void
regex_find_path (const struct GNUNET_HashCode *key,
                 struct RegexSearchContext *ctx)
{
  struct GNUNET_DHT_GetHandle *get_h;

  LOG (GNUNET_ERROR_TYPE_DEBUG,
       "Accept state found, now searching for paths to %s\n",
       GNUNET_h2s (key),
       (unsigned int) ctx->position);
  get_h = GNUNET_DHT_get_start (ctx->info->dht,    /* handle */
                                GNUNET_BLOCK_TYPE_REGEX_ACCEPT, /* type */
                                key,     /* key to search */
                                DHT_REPLICATION, /* replication level */
                                DHT_OPT | GNUNET_DHT_RO_RECORD_ROUTE,
                                NULL,       /* xquery */ // FIXME BLOOMFILTER
                                0,     /* xquery bits */ // FIXME BLOOMFILTER SIZE
                                &dht_get_string_accept_handler, ctx);
  GNUNET_break (GNUNET_OK ==
                GNUNET_CONTAINER_multihashmap_put(ctx->info->dht_get_handles,
                                                  key,
                                                  get_h,
                                                  GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE));
}

/**
 * Perform an asynchronous reverse lookup operation on the GNS.
 *
 * @param handle handle to the GNS service
 * @param zone_key zone to find a name for
 * @param root_key our zone
 * @param proc processor to call on result
 * @param proc_cls closure for @a proc
 * @return handle to the request
 */
struct GNUNET_GNS_ReverseLookupRequest*
GNUNET_GNS_reverse_lookup (struct GNUNET_GNS_Handle *handle,
                           const struct GNUNET_CRYPTO_EcdsaPublicKey *zone_key,
                           const struct GNUNET_CRYPTO_EcdsaPublicKey *root_key,
                           GNUNET_GNS_ReverseLookupResultProcessor proc,
                           void *proc_cls)
{
    /* IPC to shorten gns names, return shorten_handle */
    struct ReverseLookupMessage *rev_lookup_msg;
    struct GNUNET_GNS_ReverseLookupRequest *lr;

    if ((NULL == zone_key) || (NULL == root_key))
    {
        GNUNET_break (0);
        return NULL;
    }
    LOG (GNUNET_ERROR_TYPE_DEBUG,
         "Trying to reverse lookup in GNS\n");
    lr = GNUNET_new (struct GNUNET_GNS_ReverseLookupRequest);
    lr->gns_handle = handle;
    lr->lookup_proc = proc;
    lr->proc_cls = proc_cls;
    lr->r_id = handle->r_id_gen++;
    lr->env = GNUNET_MQ_msg (rev_lookup_msg,
                             GNUNET_MESSAGE_TYPE_GNS_REVERSE_LOOKUP);
    rev_lookup_msg->id = htonl (lr->r_id);
    rev_lookup_msg->zone_pkey = *zone_key;
    rev_lookup_msg->root_pkey = *root_key;
    GNUNET_CONTAINER_DLL_insert (handle->rev_lookup_head,
                                 handle->rev_lookup_tail,
                                 lr);
    if (NULL != handle->mq)
        GNUNET_MQ_send_copy (handle->mq,
                             lr->env);
    return lr;
}

/**
 * Handle element for iteration over the set.  Notifies the
 * iterator and sends an acknowledgement to the service.
 *
 * @param cls the `struct GNUNET_SET_Handle *`
 * @param mh the message
 */
static void
handle_copy_lazy (void *cls,
                  const struct GNUNET_MessageHeader *mh)
{
  struct GNUNET_SET_CopyLazyResponseMessage *msg;
  struct GNUNET_SET_Handle *set = cls;
  struct SetCopyRequest *req;
  struct GNUNET_SET_Handle *new_set;

  msg = (struct GNUNET_SET_CopyLazyResponseMessage *) mh;

  req = set->copy_req_head;

  if (NULL == req)
  {
    /* Service sent us unsolicited lazy copy response */
    GNUNET_break (0);
    return;
  }

  LOG (GNUNET_ERROR_TYPE_DEBUG,
       "Handling response to lazy copy\n");
  
  GNUNET_CONTAINER_DLL_remove (set->copy_req_head,
                               set->copy_req_tail,
                               req);

  
  // We pass none as operation here, since it doesn't matter when
  // cloning.
  new_set = create_internal (set->cfg, GNUNET_SET_OPERATION_NONE, &msg->cookie);

  req->cb (req->cls, new_set);

  GNUNET_free (req);
}

static void
connect_notify (void *cls, const struct GNUNET_PeerIdentity *peer)
{
  struct PeerContext *pc = cls;

  if (0 == memcmp (&pc->id, peer, sizeof (struct GNUNET_PeerIdentity)))
    return;                     /* loopback */
  GNUNET_assert (pc->connect_status == 0);
  pc->connect_status = 1;
  if (pc == &p1)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Encrypted connection established to peer `%4s'\n",
                GNUNET_i2s (peer));
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Asking core (1) for transmission to peer `%4s'\n",
                GNUNET_i2s (&p2.id));
    if (err_task != NULL)
      GNUNET_SCHEDULER_cancel (err_task);
    err_task =
        GNUNET_SCHEDULER_add_delayed (TIMEOUT, &terminate_task_error, NULL);
    start_time = GNUNET_TIME_absolute_get ();
    running = GNUNET_YES;
    measure_task =
        GNUNET_SCHEDULER_add_delayed (MEASUREMENT_LENGTH, &measurement_stop,
                                      NULL);

    GNUNET_break (NULL !=
                  (p1.nth =
                   GNUNET_CORE_notify_transmit_ready (p1.ch, GNUNET_NO,
                                                      GNUNET_CORE_PRIO_BEST_EFFORT,
                                                      TIMEOUT, &p2.id,
                                                      MESSAGESIZE,
                                                      &transmit_ready, &p1)));
  }
}

/**
 * Handler for port open requests.
 *
 * @param cls Identification of the client.
 * @param pmsg The actual message.
 */
static void
handle_port_open (void *cls,
                  const struct GNUNET_CADET_PortMessage *pmsg)
{
  struct CadetClient *c = cls;

  LOG (GNUNET_ERROR_TYPE_DEBUG,
       "Open port %s requested by client %u\n",
       GNUNET_h2s (&pmsg->port),
       c->id);
  if (NULL == c->ports)
    c->ports = GNUNET_CONTAINER_multihashmap_create (4,
                                                     GNUNET_NO);
  if (GNUNET_OK !=
      GNUNET_CONTAINER_multihashmap_put (c->ports,
                                         &pmsg->port,
                                         c,
                                         GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY))
  {
    GNUNET_break (0);
    GNUNET_SERVICE_client_drop (c->client);
    return;
  }
  /* store in global hashmap */
  /* FIXME only allow one client to have the port open,
   *       have a backup hashmap with waiting clients */
  GNUNET_CONTAINER_multihashmap_put (open_ports,
                                     &pmsg->port,
                                     c,
                                     GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
  GNUNET_CONTAINER_multihashmap_get_multiple (loose_channels,
                                              &pmsg->port,
                                              &bind_loose_channel,
                                              c);
  GNUNET_SERVICE_client_continue (c->client);
}

static void
read_call (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
{
  struct GNUNET_DISK_FileHandle *stdout_read_handle = cls;
  char buf[16];

  memset (&buf, 0, sizeof (buf));
  int bytes;

  bytes = GNUNET_DISK_file_read (stdout_read_handle, &buf, sizeof (buf));

  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "bytes is %d\n", bytes);

  if (bytes < 1)
  {
    GNUNET_break (0);
    ok = 1;
    GNUNET_SCHEDULER_cancel (die_task);
    GNUNET_SCHEDULER_add_now (&end_task, NULL);
    return;
  }

  ok = strncmp (&buf[0], test_phrase, strlen (test_phrase));
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "read %s\n", &buf[0]);
  if (0 == ok)
  {
    GNUNET_SCHEDULER_cancel (die_task);
    GNUNET_SCHEDULER_add_now (&end_task, NULL);
    return;
  }

  GNUNET_SCHEDULER_add_read_file (GNUNET_TIME_UNIT_FOREVER_REL,
                                  stdout_read_handle, &read_call,
                                  stdout_read_handle);

}

/**
 * Create a new entry in the peer list.
 *
 * @param peer identity of the new entry
 * @param hello hello message, can be NULL
 * @param is_friend is the new entry for a friend?
 * @return the new entry
 */
static struct Peer *
make_peer (const struct GNUNET_PeerIdentity *peer,
           const struct GNUNET_HELLO_Message *hello,
           int is_friend)
{
  struct Peer *ret;

  ret = GNUNET_new (struct Peer);
  ret->pid = *peer;
  ret->is_friend = is_friend;
  if (NULL != hello)
  {
    ret->hello = GNUNET_malloc (GNUNET_HELLO_size (hello));
    GNUNET_memcpy (ret->hello,
		   hello,
		   GNUNET_HELLO_size (hello));
  }
  GNUNET_break (GNUNET_OK ==
                GNUNET_CONTAINER_multipeermap_put (peers,
                                                   peer,
                                                   ret,
                                                   GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
  return ret;
}

/**
 * Store an item in the datastore.
 *
 * @param h handle to the datacache
 * @param key key to store data under
 * @param size number of bytes in data
 * @param data data to store
 * @param type type of the value
 * @param discard_time when to discard the value in any case
 * @return GNUNET_OK on success, GNUNET_SYSERR on error (full, etc.)
 */
int
GNUNET_DATACACHE_put (struct GNUNET_DATACACHE_Handle *h,
                      const GNUNET_HashCode * key, size_t size,
                      const char *data, enum GNUNET_BLOCK_Type type,
                      struct GNUNET_TIME_Absolute discard_time)
{
  uint32_t used;

  used = h->api->put (h->api->cls, key, size, data, type, discard_time);
  if (used == 0)
  {
    GNUNET_break (0);
    return GNUNET_SYSERR;
  }
  LOG (GNUNET_ERROR_TYPE_DEBUG, "Stored data under key `%s' in cache\n",
       GNUNET_h2s (key));
  GNUNET_STATISTICS_update (h->stats, gettext_noop ("# bytes stored"), size,
                            GNUNET_NO);
  GNUNET_CONTAINER_bloomfilter_add (h->filter, key);
  while (h->utilization + used > h->env.quota)
    GNUNET_assert (GNUNET_OK == h->api->del (h->api->cls));
  h->utilization += used;
  return GNUNET_OK;
}

static int
run_properties (void)
{
  struct SysmonProperty *sp;

  for (sp = sp_head; NULL != sp; sp = sp->next)
  {
      if (t_static == sp->type)
      {
          GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Running static property `%s' \n", sp->desc);
          put_property (sp);
      }
      else
      {
          if (NULL == sp->task)
          {
            GNUNET_break (0);
            continue;
          }
          sp->task_id = GNUNET_SCHEDULER_add_now (&run_property, sp);
      }
  }
  return GNUNET_OK;
}

/**
 * Convert IP address to string without DNS resolution.
 *
 * @param af address family
 * @param ip the address
 * @param ip_len number of bytes in @a ip
 * @return address as a string, NULL on error
 */
static char *
no_resolve (int af,
            const void *ip, socklen_t ip_len)
{
    char buf[INET6_ADDRSTRLEN];

    switch (af)
    {
    case AF_INET:
        if (ip_len != sizeof (struct in_addr))
            return NULL;
        if (NULL ==
                inet_ntop (AF_INET, ip, buf, sizeof (buf)))
        {
            LOG_STRERROR (GNUNET_ERROR_TYPE_WARNING,
                          "inet_ntop");
            return NULL;
        }
        break;
    case AF_INET6:
        if (ip_len != sizeof (struct in6_addr))
            return NULL;
        if (NULL ==
                inet_ntop (AF_INET6, ip, buf, sizeof (buf)))
        {
            LOG_STRERROR (GNUNET_ERROR_TYPE_WARNING,
                          "inet_ntop");
            return NULL;
        }
        break;
    default:
        GNUNET_break (0);
        return NULL;
    }
    return GNUNET_strdup (buf);
}

/**
 * Notify the tracker that a certain number of bytes of bandwidth have
 * been consumed.  Note that it is legal to consume bytes even if not
 * enough bandwidth is available (in that case,
 * #GNUNET_BANDWIDTH_tracker_get_delay may return non-zero delay values
 * even for a size of zero for a while).
 *
 * @param av tracker to update
 * @param size number of bytes consumed
 * @return #GNUNET_YES if this consumption is above the limit
 */
int
GNUNET_BANDWIDTH_tracker_consume (struct GNUNET_BANDWIDTH_Tracker *av,
                                  ssize_t size)
{
  int64_t nc;

  LOG (GNUNET_ERROR_TYPE_DEBUG,
       "Tracker %p consumes %d bytes\n",
       av,
       (int) size);
  if (size > 0)
  {
    nc = av->consumption_since_last_update__ + size;
    if (nc < av->consumption_since_last_update__)
    {
      GNUNET_break (0);
      return GNUNET_SYSERR;
    }
    av->consumption_since_last_update__ = nc;
    update_tracker (av);
    update_excess (av);
    if (av->consumption_since_last_update__ > 0)
    {
      LOG (GNUNET_ERROR_TYPE_DEBUG,
           "Tracker %p consumption %llu bytes above limit\n", av,
           (unsigned long long) av->consumption_since_last_update__);
      return GNUNET_YES;
    }
  }
  else
  {
    av->consumption_since_last_update__ += size;
    update_excess (av);
  }
  return GNUNET_NO;
}

static int
test_policy6toregex (const char *policy,
                     const char *regex)
{
  char *r;
  int ret;

  ret = 0;
  r = GNUNET_TUN_ipv6policy2regex (policy);
  if (NULL == r)
  {
    GNUNET_break (0);
    return 1;
  }
  if (0 != strcmp (regex, r))
  {
    GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
                "Expected: `%s' but got: `%s'\n",
                regex, r);
    ret = 2;
  }
  GNUNET_free (r);
  return ret;
}

int
main (int argc, char *argv[])
{
  int errCnt = 0;

  GNUNET_log_setup ("test_getopt", "WARNING", NULL);
  /* suppress output from -h, -v options */
#ifndef MINGW
  GNUNET_break (0 == CLOSE (1));
#endif
  if (0 != testMinimal ())
    errCnt++;
  if (0 != testVerbose ())
    errCnt++;
  if (0 != testVersion ())
    errCnt++;
  if (0 != testAbout ())
    errCnt++;
  if (0 != testLogOpts ())
    errCnt++;
  if (0 != testFlagNum ())
    errCnt++;
  return errCnt;
}

static void
curl_main ()
{
  fd_set rs;
  fd_set ws;
  fd_set es;
  int max;
  struct GNUNET_NETWORK_FDSet nrs;
  struct GNUNET_NETWORK_FDSet nws;
  struct GNUNET_TIME_Relative delay;
  long timeout;
  int running;
  struct CURLMsg *msg;

  max = 0;
  FD_ZERO (&rs);
  FD_ZERO (&ws);
  FD_ZERO (&es);
  curl_multi_perform (multi, &running);
  if (running == 0)
  {
    GNUNET_assert (NULL != (msg = curl_multi_info_read (multi, &running)));
    if (msg->msg == CURLMSG_DONE)
    {
      if (msg->data.result != CURLE_OK)
      {
	fprintf (stderr,
		 "%s failed at %s:%d: `%s'\n",
		 "curl_multi_perform",
		__FILE__,
		__LINE__, curl_easy_strerror (msg->data.result));
	global_ret = 1;
      }
    }
    curl_multi_remove_handle (multi, curl);
    curl_multi_cleanup (multi);
    curl_easy_cleanup (curl);
    curl = NULL;
    multi = NULL;
    if (cbc.pos != strlen ("/hello_world"))
    {
      GNUNET_break (0);
      global_ret = 2;
    }
    if (0 != strncmp ("/hello_world", cbc.buf, strlen ("/hello_world")))
    {
      GNUNET_break (0);
      global_ret = 3;
    }
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Download complete, shutting down!\n");
    do_shutdown ();
    return;
  }
  GNUNET_assert (CURLM_OK == curl_multi_fdset (multi, &rs, &ws, &es, &max));
  if ( (CURLM_OK != curl_multi_timeout (multi, &timeout)) ||
       (-1 == timeout) )
    delay = GNUNET_TIME_UNIT_SECONDS;
  else
    delay = GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MILLISECONDS, (unsigned int) timeout);
  GNUNET_NETWORK_fdset_copy_native (&nrs,
				    &rs,
				    max + 1);
  GNUNET_NETWORK_fdset_copy_native (&nws,
				    &ws,
				    max + 1);
  curl_task_id = GNUNET_SCHEDULER_add_select (GNUNET_SCHEDULER_PRIORITY_DEFAULT,
					      delay,
					      &nrs,
					      &nws,
					      &curl_task,
					      NULL);
}

void
GNUNET_ATS_TEST_logging_write_to_file (struct LoggingHandle *l,
    char *experiment_name, int plots)
{
  struct GNUNET_DISK_FileHandle *f[l->num_slaves];
  struct GNUNET_DISK_FileHandle *f_m;
  char *tmp_exp_name;
  char *filename_master;
  char *filename_slaves[l->num_slaves];
  char *data;
  struct PeerLoggingTimestep *cur_lt;
  struct PartnerLoggingTimestep *plt;
  struct GNUNET_TIME_Absolute timestamp;
  int c_m;
  int c_s;


  timestamp = GNUNET_TIME_absolute_get();

  tmp_exp_name = experiment_name;
  for (c_m = 0; c_m < l->num_masters; c_m++)
  {
    GNUNET_asprintf (&filename_master, "%s_%llu_master%u_%s",
        experiment_name, timestamp.abs_value_us, c_m, l->name);
    fprintf (stderr, "Writing data for master %u to file `%s'\n",
        c_m,filename_master);

    f_m = GNUNET_DISK_file_open (filename_master,
        GNUNET_DISK_OPEN_WRITE | GNUNET_DISK_OPEN_CREATE,
        GNUNET_DISK_PERM_USER_READ | GNUNET_DISK_PERM_USER_WRITE);
    if (NULL == f_m)
    {
      GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "Cannot open log file `%s'\n", filename_master);
      GNUNET_free (filename_master);
      return;
    }

    GNUNET_asprintf (&data, "# master %u; experiment : %s\n"
        "timestamp; timestamp delta; #messages sent; #bytes sent; #throughput sent; #messages received; #bytes received; #throughput received; \n" ,
        c_m,  experiment_name);
    if (GNUNET_SYSERR == GNUNET_DISK_file_write(f_m, data, strlen(data)))
      GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
          "Cannot write data to log file `%s'\n",filename_master);
    GNUNET_free (data);

    for (c_s = 0; c_s < l->lp[c_m].peer->num_partners; c_s++)
    {
      GNUNET_asprintf (&filename_slaves[c_s], "%s_%llu_master%u_slave_%u_%s",
          tmp_exp_name, timestamp.abs_value_us, c_m, c_s, l->name);

      fprintf (stderr, "Writing data for master %u slave %u to file `%s'\n",
          c_m, c_s, filename_slaves[c_s]);

      f[c_s] = GNUNET_DISK_file_open (filename_slaves[c_s],
          GNUNET_DISK_OPEN_WRITE | GNUNET_DISK_OPEN_CREATE,
          GNUNET_DISK_PERM_USER_READ | GNUNET_DISK_PERM_USER_WRITE);
      if (NULL == f[c_s])
      {
        GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "Cannot open log file `%s'\n", filename_slaves[c_s]);
        GNUNET_free (filename_slaves[c_s]);
        GNUNET_break (GNUNET_OK == GNUNET_DISK_file_close(f_m));
        GNUNET_free (filename_master);
        return;
      }

      /* Header */
      GNUNET_asprintf (&data, "# master %u; slave %u ; experiment : %s\n"
          "timestamp; timestamp delta; #messages sent; #bytes sent; #throughput sent; #messages received; #bytes received; #throughput received; " \
          "rtt; bw in; bw out; ats_cost_lan; ats_cost_wlan; ats_delay; ats_distance; ats_network_type; ats_utilization_up ;ats_utilization_down;" \
          "pref bandwidth; pref delay\n",
          c_m, c_s, experiment_name);
      if (GNUNET_SYSERR == GNUNET_DISK_file_write(f[c_s], data, strlen(data)))
        GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
            "Cannot write data to log file `%s'\n",filename_slaves[c_s]);
      GNUNET_free (data);
    }

    for (cur_lt = l->lp[c_m].head; NULL != cur_lt; cur_lt = cur_lt->next)
    {
      if (l->verbose)
        fprintf (stderr,
           "Master [%u]: timestamp %llu %llu ; %u %u %u ; %u %u %u\n",
           l->lp[c_m].peer->no,
           (long long unsigned int) cur_lt->timestamp.abs_value_us,
           (long long unsigned int) GNUNET_TIME_absolute_get_difference(l->lp[c_m].start,
               cur_lt->timestamp).rel_value_us / 1000,
           cur_lt->total_messages_sent,
           cur_lt->total_bytes_sent,
           cur_lt->total_throughput_send,
           cur_lt->total_messages_received,
           cur_lt->total_bytes_received,
           cur_lt->total_throughput_recv);

      /* Assembling master string */
      GNUNET_asprintf (&data, "%llu;%llu;%u;%u;%u;%u;%u;%u;\n",
          (long long unsigned int) cur_lt->timestamp.abs_value_us,
          (long long unsigned int) GNUNET_TIME_absolute_get_difference(l->lp[c_m].start,
              cur_lt->timestamp).rel_value_us / 1000,
          cur_lt->total_messages_sent,
          cur_lt->total_bytes_sent,
//.........这里部分代码省略.........