/*
This file is part of GNUnet.
- (C) 2009, 2010, 2011 Christian Grothoff (and other contributing authors)
+ (C) 2009, 2010, 2011, 2012 Christian Grothoff (and other contributing authors)
GNUnet is free software; you can redistribute it and/or modify
it under the terms of the GNU General Public License as published
*/
#define MAX_RESULTS (100 * 1024)
+/**
+ * Collect an instane number of statistics? May cause excessive IPC.
+ */
+#define INSANE_STATISTICS GNUNET_NO
+
/**
* An active request.
*/
/**
* Array of hash codes of replies we've already seen.
*/
- GNUNET_HashCode *replies_seen;
+ struct GNUNET_HashCode *replies_seen;
/**
* Bloomfilter masking replies we've already seen.
struct GSF_PendingRequest *
GSF_pending_request_create_ (enum GSF_PendingRequestOptions options,
enum GNUNET_BLOCK_Type type,
- const GNUNET_HashCode * query,
- const GNUNET_HashCode * namespace,
+ const struct GNUNET_HashCode * query,
+ const struct GNUNET_HashCode * namespace,
const struct GNUNET_PeerIdentity *target,
const char *bf_data, size_t bf_size,
uint32_t mingle, uint32_t anonymity_level,
uint32_t priority, int32_t ttl,
GNUNET_PEER_Id sender_pid,
GNUNET_PEER_Id origin_pid,
- const GNUNET_HashCode * replies_seen,
+ const struct GNUNET_HashCode * replies_seen,
unsigned int replies_seen_count,
GSF_PendingRequestReplyHandler rh, void *rh_cls)
{
struct GSF_PendingRequest *pr;
struct GSF_PendingRequest *dpr;
+ size_t extra;
+ struct GNUNET_HashCode *eptr;
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
"Creating request handle for `%s' of type %d\n",
GNUNET_h2s (query), type);
+#if INSANE_STATISTICS
GNUNET_STATISTICS_update (GSF_stats,
gettext_noop ("# Pending requests created"), 1,
GNUNET_NO);
- pr = GNUNET_malloc (sizeof (struct GSF_PendingRequest));
+#endif
+ extra = 0;
+ if (GNUNET_BLOCK_TYPE_FS_SBLOCK == type)
+ extra += sizeof (struct GNUNET_HashCode);
+ if (NULL != target)
+ extra += sizeof (struct GNUNET_PeerIdentity);
+ pr = GNUNET_malloc (sizeof (struct GSF_PendingRequest) + extra);
pr->local_result_offset =
GNUNET_CRYPTO_random_u64 (GNUNET_CRYPTO_QUALITY_WEAK, UINT64_MAX);
pr->public_data.query = *query;
+ eptr = (struct GNUNET_HashCode *) &pr[1];
if (GNUNET_BLOCK_TYPE_FS_SBLOCK == type)
{
- GNUNET_assert (NULL != namespace);
- pr->public_data.namespace = *namespace;
+ GNUNET_assert (NULL != namespace);
+ pr->public_data.namespace = eptr;
+ memcpy (eptr, namespace, sizeof (struct GNUNET_HashCode));
+ eptr++;
}
if (NULL != target)
{
- pr->public_data.target = *target;
- pr->public_data.has_target = GNUNET_YES;
+ pr->public_data.target = (struct GNUNET_PeerIdentity *) eptr;
+ memcpy (eptr, target, sizeof (struct GNUNET_PeerIdentity));
}
pr->public_data.anonymity_level = anonymity_level;
pr->public_data.priority = priority;
{
pr->replies_seen_size = replies_seen_count;
pr->replies_seen =
- GNUNET_malloc (sizeof (GNUNET_HashCode) * pr->replies_seen_size);
+ GNUNET_malloc (sizeof (struct GNUNET_HashCode) * pr->replies_seen_size);
memcpy (pr->replies_seen, replies_seen,
- replies_seen_count * sizeof (GNUNET_HashCode));
+ replies_seen_count * sizeof (struct GNUNET_HashCode));
pr->replies_seen_count = replies_seen_count;
}
if (NULL != bf_data)
{
refresh_bloomfilter (pr);
}
- GNUNET_CONTAINER_multihashmap_put (pr_map, query, pr,
+ GNUNET_CONTAINER_multihashmap_put (pr_map,
+ &pr->public_data.query, pr,
GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
if (0 == (options & GSF_PRO_REQUEST_NEVER_EXPIRES))
{
if ((pra->public_data.type != prb->public_data.type) ||
(0 !=
memcmp (&pra->public_data.query, &prb->public_data.query,
- sizeof (GNUNET_HashCode))) ||
+ sizeof (struct GNUNET_HashCode))) ||
((pra->public_data.type == GNUNET_BLOCK_TYPE_FS_SBLOCK) &&
(0 !=
- memcmp (&pra->public_data.namespace, &prb->public_data.namespace,
- sizeof (GNUNET_HashCode)))))
+ memcmp (pra->public_data.namespace,
+ prb->public_data.namespace,
+ sizeof (struct GNUNET_HashCode)))))
return GNUNET_NO;
return GNUNET_OK;
}
*/
void
GSF_pending_request_update_ (struct GSF_PendingRequest *pr,
- const GNUNET_HashCode * replies_seen,
+ const struct GNUNET_HashCode * replies_seen,
unsigned int replies_seen_count)
{
unsigned int i;
- GNUNET_HashCode mhash;
+ struct GNUNET_HashCode mhash;
if (replies_seen_count + pr->replies_seen_count < pr->replies_seen_count)
return; /* integer overflow */
GNUNET_array_grow (pr->replies_seen, pr->replies_seen_size,
replies_seen_count + pr->replies_seen_count);
memcpy (&pr->replies_seen[pr->replies_seen_count], replies_seen,
- sizeof (GNUNET_HashCode) * replies_seen_count);
+ sizeof (struct GNUNET_HashCode) * replies_seen_count);
pr->replies_seen_count += replies_seen_count;
refresh_bloomfilter (pr);
}
{
char lbuf[GNUNET_SERVER_MAX_MESSAGE_SIZE];
struct GetMessage *gm;
- GNUNET_HashCode *ext;
+ struct GNUNET_HashCode *ext;
size_t msize;
unsigned int k;
uint32_t bm;
bm |= GET_MESSAGE_BIT_SKS_NAMESPACE;
k++;
}
- if (GNUNET_YES == pr->public_data.has_target)
+ if (NULL != pr->public_data.target)
{
bm |= GET_MESSAGE_BIT_TRANSMIT_TO;
k++;
}
bf_size = GNUNET_CONTAINER_bloomfilter_get_size (pr->bf);
- msize = sizeof (struct GetMessage) + bf_size + k * sizeof (GNUNET_HashCode);
+ msize = sizeof (struct GetMessage) + bf_size + k * sizeof (struct GNUNET_HashCode);
GNUNET_assert (msize < GNUNET_SERVER_MAX_MESSAGE_SIZE);
if (buf_size < msize)
return msize;
else
prio = 0;
pr->public_data.priority -= prio;
+ pr->public_data.num_transmissions++;
+ pr->public_data.respect_offered += prio;
gm->priority = htonl (prio);
now = GNUNET_TIME_absolute_get ();
ttl = (int64_t) (pr->public_data.ttl.abs_value - now.abs_value);
gm->filter_mutator = htonl (pr->mingle);
gm->hash_bitmap = htonl (bm);
gm->query = pr->public_data.query;
- ext = (GNUNET_HashCode *) & gm[1];
+ ext = (struct GNUNET_HashCode *) & gm[1];
k = 0;
if (!do_route)
GNUNET_PEER_resolve (pr->sender_pid,
(struct GNUNET_PeerIdentity *) &ext[k++]);
if (GNUNET_BLOCK_TYPE_FS_SBLOCK == pr->public_data.type)
- memcpy (&ext[k++], &pr->public_data.namespace, sizeof (GNUNET_HashCode));
- if (GNUNET_YES == pr->public_data.has_target)
- ext[k++] = pr->public_data.target.hashPubKey;
+ memcpy (&ext[k++], pr->public_data.namespace, sizeof (struct GNUNET_HashCode));
+ if (NULL != pr->public_data.target)
+ memcpy (&ext[k++],
+ pr->public_data.target,
+ sizeof (struct GNUNET_PeerIdentity));
if (pr->bf != NULL)
GNUNET_assert (GNUNET_SYSERR !=
GNUNET_CONTAINER_bloomfilter_get_raw_data (pr->bf,
* @return GNUNET_YES (we should continue to iterate)
*/
static int
-clean_request (void *cls, const GNUNET_HashCode * key, void *value)
+clean_request (void *cls, const struct GNUNET_HashCode * key, void *value)
{
struct GSF_PendingRequest *pr = value;
GSF_LocalLookupContinuation cont;
* @return GNUNET_YES (we should continue to iterate)
*/
static int
-process_reply (void *cls, const GNUNET_HashCode * key, void *value)
+process_reply (void *cls, const struct GNUNET_HashCode * key, void *value)
{
struct ProcessReplyClosure *prq = cls;
struct GSF_PendingRequest *pr = value;
- GNUNET_HashCode chash;
+ struct GNUNET_HashCode chash;
struct GNUNET_TIME_Absolute last_transmission;
if (NULL == pr->rh)
GNUNET_NO);
prq->eval =
GNUNET_BLOCK_evaluate (GSF_block_ctx, prq->type, key, &pr->bf, pr->mingle,
- &pr->public_data.namespace,
+ pr->public_data.namespace,
(prq->type ==
GNUNET_BLOCK_TYPE_FS_SBLOCK) ?
- sizeof (GNUNET_HashCode) : 0, prq->data,
+ sizeof (struct GNUNET_HashCode) : 0, prq->data,
prq->size);
switch (prq->eval)
{
last_transmission, prq->type, prq->data, prq->size);
return GNUNET_YES;
case GNUNET_BLOCK_EVALUATION_OK_DUPLICATE:
+#if INSANE_STATISTICS
GNUNET_STATISTICS_update (GSF_stats,
gettext_noop
("# duplicate replies discarded (bloomfilter)"),
1, GNUNET_NO);
+#endif
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
"Duplicate response, discarding.\n");
return GNUNET_YES; /* duplicate */
if (min_expiration.abs_value > 0)
{
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Asking to stop migration for %llu ms because datastore is full\n",
- (unsigned long long) GNUNET_TIME_absolute_get_remaining (min_expiration).rel_value);
+ "Asking to stop migration for %s because datastore is full\n",
+ GNUNET_STRINGS_relative_time_to_string (GNUNET_TIME_absolute_get_remaining (min_expiration), GNUNET_YES));
GSF_block_peer_migration_ (cp, min_expiration);
}
else
ppd->migration_delay.rel_value);
ppd->migration_delay = GNUNET_TIME_relative_multiply (ppd->migration_delay, 2);
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Replicated content already exists locally, asking to stop migration for %llu ms\n",
- (unsigned long long) mig_pause.rel_value);
+ "Replicated content already exists locally, asking to stop migration for %s\n",
+ GNUNET_STRINGS_relative_time_to_string (mig_pause, GNUNET_YES));
GSF_block_peer_migration_ (cp, GNUNET_TIME_relative_to_absolute (mig_pause));
}
}
*/
static void
handle_dht_reply (void *cls, struct GNUNET_TIME_Absolute exp,
- const GNUNET_HashCode * key,
+ const struct GNUNET_HashCode * key,
const struct GNUNET_PeerIdentity *get_path,
unsigned int get_path_length,
const struct GNUNET_PeerIdentity *put_path,
const void *xquery;
size_t xquery_size;
struct GNUNET_PeerIdentity pi;
- char buf[sizeof (GNUNET_HashCode) * 2];
+ char buf[sizeof (struct GNUNET_HashCode) * 2] GNUNET_ALIGN;
if (0 != pr->public_data.anonymity_level)
return;
if (GNUNET_BLOCK_TYPE_FS_SBLOCK == pr->public_data.type)
{
xquery = buf;
- memcpy (buf, &pr->public_data.namespace, sizeof (GNUNET_HashCode));
- xquery_size = sizeof (GNUNET_HashCode);
+ memcpy (buf, pr->public_data.namespace, sizeof (struct GNUNET_HashCode));
+ xquery_size = sizeof (struct GNUNET_HashCode);
}
if (0 != (pr->public_data.options & GSF_PRO_FORWARD_ONLY))
{
xquery_size += sizeof (struct GNUNET_PeerIdentity);
}
pr->gh =
- GNUNET_DHT_get_start (GSF_dht, GNUNET_TIME_UNIT_FOREVER_REL,
+ GNUNET_DHT_get_start (GSF_dht,
pr->public_data.type, &pr->public_data.query,
5 /* DEFAULT_GET_REPLICATION */ ,
GNUNET_DHT_RO_DEMULTIPLEX_EVERYWHERE,
struct GSF_PendingRequest *pr = cls;
GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
- _("Datastore lookup already took %llu ms!\n"),
- (unsigned long long)
- GNUNET_TIME_absolute_get_duration (pr->qe_start).rel_value);
+ _("Datastore lookup already took %s!\n"),
+ GNUNET_STRINGS_relative_time_to_string (GNUNET_TIME_absolute_get_duration (pr->qe_start), GNUNET_YES));
pr->warn_task =
GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_MINUTES, &warn_delay_task,
pr);
struct GSF_PendingRequest *pr = cls;
GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
- _("On-demand lookup already took %llu ms!\n"),
- (unsigned long long)
- GNUNET_TIME_absolute_get_duration (pr->qe_start).rel_value);
+ _("On-demand lookup already took %s!\n"),
+ GNUNET_STRINGS_relative_time_to_string (GNUNET_TIME_absolute_get_duration (pr->qe_start), GNUNET_YES));
pr->warn_task =
GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_MINUTES,
&odc_warn_delay_task, pr);
* maybe 0 if no unique identifier is available
*/
static void
-process_local_reply (void *cls, const GNUNET_HashCode * key, size_t size,
+process_local_reply (void *cls, const struct GNUNET_HashCode * key, size_t size,
const void *data, enum GNUNET_BLOCK_Type type,
uint32_t priority, uint32_t anonymity,
struct GNUNET_TIME_Absolute expiration, uint64_t uid)
struct GSF_PendingRequest *pr = cls;
GSF_LocalLookupContinuation cont;
struct ProcessReplyClosure prq;
- GNUNET_HashCode query;
+ struct GNUNET_HashCode query;
unsigned int old_rf;
GNUNET_SCHEDULER_cancel (pr->warn_task);
pr->qe = NULL;
if (NULL == key)
{
+#if INSANE_STATISTICS
GNUNET_STATISTICS_update (GSF_stats,
gettext_noop
("# Datastore lookups concluded (no results)"),
1, GNUNET_NO);
+#endif
}
if (GNUNET_NO == pr->have_first_uid)
{
{
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG | GNUNET_ERROR_TYPE_BULK,
"No further local responses available.\n");
+#if INSANE_STATISTICS
if ((pr->public_data.type == GNUNET_BLOCK_TYPE_FS_DBLOCK) ||
(pr->public_data.type == GNUNET_BLOCK_TYPE_FS_IBLOCK))
GNUNET_STATISTICS_update (GSF_stats,
gettext_noop
("# requested DBLOCK or IBLOCK not found"), 1,
GNUNET_NO);
+#endif
goto check_error_and_continue;
}
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
pr->warn_task =
GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_MINUTES, &warn_delay_task,
pr);
+#if INSANE_STATISTICS
GNUNET_STATISTICS_update (GSF_stats,
gettext_noop ("# Datastore lookups initiated"), 1,
GNUNET_NO);
+#endif
pr->qe =
GNUNET_DATASTORE_get_key (GSF_dsh, pr->local_result_offset++,
&pr->public_data.query,
size_t dsize;
enum GNUNET_BLOCK_Type type;
struct GNUNET_TIME_Absolute expiration;
- GNUNET_HashCode query;
+ struct GNUNET_HashCode query;
struct ProcessReplyClosure prq;
struct GNUNET_TIME_Relative block_time;
double putl;
GNUNET_NO);
/* now, lookup 'query' */
prq.data = (const void *) &put[1];
- if (NULL != cp)
- prq.sender = cp;
- else
- prq.sender = NULL;
+ prq.sender = cp;
prq.size = dsize;
prq.type = type;
prq.expiration = expiration;
GSF_connected_peer_change_preference_ (cp,
CONTENT_BANDWIDTH_VALUE +
1000 * prq.priority);
- GSF_get_peer_performance_data_ (cp)->trust += prq.priority;
+ GSF_get_peer_performance_data_ (cp)->respect += prq.priority;
}
if ((GNUNET_YES == active_to_migration) &&
+ (NULL != cp) &&
(GNUNET_NO == test_put_load_too_high (prq.priority)))
{
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
put_migration_continuation (pmc, GNUNET_SYSERR, GNUNET_TIME_UNIT_ZERO_ABS, NULL);
}
}
- else
+ else if (NULL != cp)
{
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
"Choosing not to keep content `%s' (%d/%d)\n",
test_put_load_too_high (prq.priority));
}
putl = GNUNET_LOAD_get_load (datastore_put_load);
- if ((NULL != (cp = prq.sender)) && (GNUNET_NO == prq.request_found) &&
- ((GNUNET_YES != active_to_migration) ||
- (putl > 2.5 * (1 + prq.priority))))
+ if ( (NULL != cp) &&
+ (GNUNET_NO == prq.request_found) &&
+ ( (GNUNET_YES != active_to_migration) ||
+ (putl > 2.5 * (1 + prq.priority)) ) )
{
if (GNUNET_YES != active_to_migration)
putl = 1.0 + GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK, 5);
void
GSF_pending_request_init_ ()
{
- unsigned long long bps;
+ unsigned long long dqs;
if (GNUNET_OK !=
GNUNET_CONFIGURATION_get_value_number (GSF_cfg, "fs",
"MAX_PENDING_REQUESTS",
&max_pending_requests))
{
- GNUNET_log (GNUNET_ERROR_TYPE_INFO,
- _
- ("Configuration fails to specify `%s', assuming default value."),
- "MAX_PENDING_REQUESTS");
+ GNUNET_log_config_missing (GNUNET_ERROR_TYPE_INFO,
+ "fs", "MAX_PENDING_REQUESTS");
}
if (GNUNET_OK !=
- GNUNET_CONFIGURATION_get_value_size (GSF_cfg, "ats", "WAN_QUOTA_OUT",
- &bps))
+ GNUNET_CONFIGURATION_get_value_size (GSF_cfg, "fs", "DATASTORE_QUEUE_SIZE",
+ &dqs))
{
- GNUNET_log (GNUNET_ERROR_TYPE_INFO,
- _
- ("Configuration fails to specify `%s', assuming default value."),
- "WAN_QUOTA_OUT");
- bps = 65536;
+ GNUNET_log_config_missing (GNUNET_ERROR_TYPE_INFO,
+ "fs", "DATASTORE_QUEUE_SIZE");
+ dqs = 1024;
}
- /* queue size should be #queries we can have pending and satisfy within
- * a carry interval: */
- datastore_queue_size =
- bps * GNUNET_CONSTANTS_MAX_BANDWIDTH_CARRY_S / DBLOCK_SIZE;
+ datastore_queue_size = (unsigned int) dqs;
active_to_migration =
GNUNET_CONFIGURATION_get_value_yesno (GSF_cfg, "FS", "CONTENT_CACHING");
datastore_put_load = GNUNET_LOAD_value_init (DATASTORE_LOAD_AUTODECLINE);
- pr_map = GNUNET_CONTAINER_multihashmap_create (32 * 1024);
+ pr_map = GNUNET_CONTAINER_multihashmap_create (32 * 1024, GNUNET_YES);
requests_by_expiration_heap =
GNUNET_CONTAINER_heap_create (GNUNET_CONTAINER_HEAP_ORDER_MIN);
}
projects / oweals / gnunet.git / blobdiff