/*
This file is part of GNUnet.
- (C) 2009-2013 Christian Grothoff (and other contributing authors)
+ Copyright (C) 2009-2013 GNUnet e.V.
GNUnet is free software; you can redistribute it and/or modify
it under the terms of the GNU General Public License as published
You should have received a copy of the GNU General Public License
along with GNUnet; see the file COPYING. If not, write to the
- Free Software Foundation, Inc., 59 Temple Place - Suite 330,
- Boston, MA 02111-1307, USA.
+ Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor,
+ Boston, MA 02110-1301, USA.
*/
/**
#include "gnunet-service-fs_indexing.h"
#include "gnunet-service-fs_pe.h"
#include "gnunet-service-fs_pr.h"
-#include "gnunet-service-fs_mesh.h"
+#include "gnunet-service-fs_cadet.h"
/**
/**
* Maximum size of the datastore queue for P2P operations. Needs to
- * be large enough to queue MAX_QUEUE_PER_PEER operations for roughly
+ * be large enough to queue #MAX_QUEUE_PER_PEER operations for roughly
* the number of active (connected) peers.
*/
#define MAX_DATASTORE_QUEUE (16 * MAX_QUEUE_PER_PEER)
#define INSANE_STATISTICS GNUNET_NO
/**
- * If obtaining a block via mesh fails, how often do we retry it before
+ * If obtaining a block via cadet fails, how often do we retry it before
* giving up for good (and sticking to non-anonymous transfer)?
*/
-#define MESH_RETRY_MAX 3
+#define CADET_RETRY_MAX 3
/**
struct GNUNET_HashCode *replies_seen;
/**
- * Bloomfilter masking replies we've already seen.
+ * Block group for filtering replies we've already seen.
*/
- struct GNUNET_CONTAINER_BloomFilter *bf;
+ struct GNUNET_BLOCK_Group *bg;
/**
* Entry for this pending request in the expiration heap, or NULL.
struct GNUNET_DHT_GetHandle *gh;
/**
- * Mesh request handle for this request (or NULL for none).
+ * Cadet request handle for this request (or NULL for none).
*/
- struct GSF_MeshRequest *mesh_request;
+ struct GSF_CadetRequest *cadet_request;
/**
* Function to call upon completion of the local get
GSF_LocalLookupContinuation llc_cont;
/**
- * Closure for llc_cont.
+ * Closure for @e llc_cont.
*/
void *llc_cont_cls;
/**
* Task that warns us if the local datastore lookup takes too long.
*/
- GNUNET_SCHEDULER_TaskIdentifier warn_task;
+ struct GNUNET_SCHEDULER_Task * warn_task;
/**
* Current offset for querying our local datastore for results.
uint64_t first_uid;
/**
- * How often have we retried this request via 'mesh'?
+ * How often have we retried this request via 'cadet'?
* (used to bound overall retries).
*/
- unsigned int mesh_retry_count;
+ unsigned int cadet_retry_count;
/**
* Number of valid entries in the 'replies_seen' array.
*/
unsigned int replies_seen_size;
- /**
- * Mingle value we currently use for the bf.
- */
- uint32_t mingle;
-
/**
* Do we have a first UID yet?
*/
* fresh one of minimal size without problems) OR if our peer is the
* initiator (in which case we may resize to larger than mimimum size).
*
+ * @param type type of the request
* @param pr request for which the BF is to be recomputed
*/
static void
-refresh_bloomfilter (struct GSF_PendingRequest *pr)
+refresh_bloomfilter (enum GNUNET_BLOCK_Type type,
+ struct GSF_PendingRequest *pr)
{
- if (pr->bf != NULL)
- GNUNET_CONTAINER_bloomfilter_free (pr->bf);
- pr->mingle =
- GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK, UINT32_MAX);
- pr->bf =
- GNUNET_BLOCK_construct_bloomfilter (pr->mingle, pr->replies_seen,
- pr->replies_seen_count);
+ if (NULL != pr->bg)
+ {
+ GNUNET_BLOCK_group_destroy (pr->bg);
+ pr->bg = NULL;
+ }
+ if (GNUNET_BLOCK_TYPE_FS_UBLOCK != type)
+ return; /* no need */
+ pr->bg
+ = GNUNET_BLOCK_group_create (GSF_block_ctx,
+ type,
+ GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
+ UINT32_MAX),
+ NULL,
+ 0,
+ "seen-set-size",
+ pr->replies_seen_count,
+ NULL);
+ if (NULL == pr->bg)
+ return;
+ GNUNET_break (GNUNET_OK ==
+ GNUNET_BLOCK_group_set_seen (pr->bg,
+ pr->replies_seen,
+ pr->replies_seen_count));
}
enum GNUNET_BLOCK_Type type,
const struct GNUNET_HashCode *query,
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,
+ 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 struct GNUNET_HashCode *replies_seen,
unsigned int replies_seen_count,
- GSF_PendingRequestReplyHandler rh, void *rh_cls)
+ GSF_PendingRequestReplyHandler rh,
+ void *rh_cls)
{
struct GSF_PendingRequest *pr;
struct GSF_PendingRequest *dpr;
if (NULL != target)
{
pr->public_data.target = (struct GNUNET_PeerIdentity *) eptr;
- memcpy (eptr, target, sizeof (struct GNUNET_PeerIdentity));
+ GNUNET_memcpy (eptr,
+ target,
+ sizeof (struct GNUNET_PeerIdentity));
}
pr->public_data.anonymity_level = anonymity_level;
pr->public_data.priority = priority;
if (replies_seen_count > 0)
{
pr->replies_seen_size = replies_seen_count;
- pr->replies_seen =
- GNUNET_malloc (sizeof (struct GNUNET_HashCode) * pr->replies_seen_size);
- memcpy (pr->replies_seen, replies_seen,
- replies_seen_count * sizeof (struct GNUNET_HashCode));
+ pr->replies_seen = GNUNET_new_array (pr->replies_seen_size,
+ struct GNUNET_HashCode);
+ GNUNET_memcpy (pr->replies_seen,
+ replies_seen,
+ replies_seen_count * sizeof (struct GNUNET_HashCode));
pr->replies_seen_count = replies_seen_count;
}
- if (NULL != bf_data)
+ if ( (NULL != bf_data) &&
+ (GNUNET_BLOCK_TYPE_FS_UBLOCK == pr->public_data.type) )
{
- pr->bf =
- GNUNET_CONTAINER_bloomfilter_init (bf_data, bf_size,
- GNUNET_CONSTANTS_BLOOMFILTER_K);
- pr->mingle = mingle;
+ pr->bg
+ = GNUNET_BLOCK_group_create (GSF_block_ctx,
+ pr->public_data.type,
+ mingle,
+ bf_data,
+ bf_size,
+ "seen-set-size",
+ 0,
+ NULL);
}
else if ((replies_seen_count > 0) &&
(0 != (options & GSF_PRO_BLOOMFILTER_FULL_REFRESH)))
{
- refresh_bloomfilter (pr);
+ refresh_bloomfilter (pr->public_data.type,
+ pr);
}
GNUNET_CONTAINER_multihashmap_put (pr_map,
- &pr->public_data.query, pr,
+ &pr->public_data.query,
+ pr,
GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
if (0 == (options & GSF_PRO_REQUEST_NEVER_EXPIRES))
{
pr->hnode =
- GNUNET_CONTAINER_heap_insert (requests_by_expiration_heap, pr,
+ GNUNET_CONTAINER_heap_insert (requests_by_expiration_heap,
+ pr,
pr->public_data.ttl.abs_value_us);
/* make sure we don't track too many requests */
while (GNUNET_CONTAINER_heap_get_size (requests_by_expiration_heap) >
max_pending_requests)
{
dpr = GNUNET_CONTAINER_heap_peek (requests_by_expiration_heap);
- GNUNET_assert (dpr != NULL);
+ GNUNET_assert (NULL != dpr);
if (pr == dpr)
break; /* let the request live briefly... */
if (NULL != dpr->rh)
- dpr->rh (dpr->rh_cls, GNUNET_BLOCK_EVALUATION_REQUEST_VALID, dpr,
- UINT32_MAX, GNUNET_TIME_UNIT_FOREVER_ABS, GNUNET_TIME_UNIT_FOREVER_ABS,
- GNUNET_BLOCK_TYPE_ANY, NULL, 0);
- GSF_pending_request_cancel_ (dpr, GNUNET_YES);
+ dpr->rh (dpr->rh_cls,
+ GNUNET_BLOCK_EVALUATION_REQUEST_VALID,
+ dpr,
+ UINT32_MAX,
+ GNUNET_TIME_UNIT_FOREVER_ABS,
+ GNUNET_TIME_UNIT_FOREVER_ABS,
+ GNUNET_BLOCK_TYPE_ANY,
+ NULL,
+ 0);
+ GSF_pending_request_cancel_ (dpr,
+ GNUNET_YES);
}
}
GNUNET_STATISTICS_update (GSF_stats,
GSF_pending_request_is_compatible_ (struct GSF_PendingRequest *pra,
struct GSF_PendingRequest *prb)
{
- if ((pra->public_data.type != prb->public_data.type) ||
- (0 !=
- memcmp (&pra->public_data.query, &prb->public_data.query,
- sizeof (struct GNUNET_HashCode))))
+ if ( (pra->public_data.type != prb->public_data.type) ||
+ (0 != memcmp (&pra->public_data.query,
+ &prb->public_data.query,
+ sizeof (struct GNUNET_HashCode))))
return GNUNET_NO;
return GNUNET_OK;
}
-
/**
* Update a given pending request with additional replies
* that have been seen.
const struct GNUNET_HashCode * replies_seen,
unsigned int replies_seen_count)
{
- unsigned int i;
- struct GNUNET_HashCode mhash;
-
if (replies_seen_count + pr->replies_seen_count < pr->replies_seen_count)
return; /* integer overflow */
if (0 != (pr->public_data.options & GSF_PRO_BLOOMFILTER_FULL_REFRESH))
{
/* we're responsible for the BF, full refresh */
if (replies_seen_count + pr->replies_seen_count > pr->replies_seen_size)
- GNUNET_array_grow (pr->replies_seen, pr->replies_seen_size,
+ 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 (struct GNUNET_HashCode) * replies_seen_count);
+ GNUNET_memcpy (&pr->replies_seen[pr->replies_seen_count],
+ replies_seen,
+ sizeof (struct GNUNET_HashCode) * replies_seen_count);
pr->replies_seen_count += replies_seen_count;
- refresh_bloomfilter (pr);
+ refresh_bloomfilter (pr->public_data.type,
+ pr);
}
else
{
- if (NULL == pr->bf)
+ if (NULL == pr->bg)
{
/* we're not the initiator, but the initiator did not give us
* any bloom-filter, so we need to create one on-the-fly */
- pr->mingle =
- GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK, UINT32_MAX);
- pr->bf =
- GNUNET_BLOCK_construct_bloomfilter (pr->mingle, replies_seen,
- replies_seen_count);
+ refresh_bloomfilter (pr->public_data.type,
+ pr);
}
else
{
- for (i = 0; i < pr->replies_seen_count; i++)
- {
- GNUNET_BLOCK_mingle_hash (&replies_seen[i], pr->mingle, &mhash);
- GNUNET_CONTAINER_bloomfilter_add (pr->bf, &mhash);
- }
+ GNUNET_break (GNUNET_OK ==
+ GNUNET_BLOCK_group_set_seen (pr->bg,
+ replies_seen,
+ pr->replies_seen_count));
}
}
if (NULL != pr->gh)
/**
* Generate the message corresponding to the given pending request for
- * transmission to other peers (or at least determine its size).
+ * transmission to other peers.
*
* @param pr request to generate the message for
- * @param buf_size number of bytes available in @a buf
- * @param buf where to copy the message (can be NULL)
- * @return number of bytes needed (if `>` @a buf_size) or used
+ * @return envelope with the request message
*/
-size_t
-GSF_pending_request_get_message_ (struct GSF_PendingRequest *pr,
- size_t buf_size, void *buf)
+struct GNUNET_MQ_Envelope *
+GSF_pending_request_get_message_ (struct GSF_PendingRequest *pr)
{
- char lbuf[GNUNET_SERVER_MAX_MESSAGE_SIZE];
+ struct GNUNET_MQ_Envelope *env;
struct GetMessage *gm;
- struct GNUNET_HashCode *ext;
- size_t msize;
+ struct GNUNET_PeerIdentity *ext;
unsigned int k;
uint32_t bm;
uint32_t prio;
struct GNUNET_TIME_Absolute now;
int64_t ttl;
int do_route;
+ void *bf_data;
+ uint32_t bf_nonce;
- if (buf_size > 0)
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Building request message for `%s' of type %d\n",
- GNUNET_h2s (&pr->public_data.query), pr->public_data.type);
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Building request message for `%s' of type %d\n",
+ GNUNET_h2s (&pr->public_data.query),
+ pr->public_data.type);
k = 0;
bm = 0;
do_route = (0 == (pr->public_data.options & GSF_PRO_FORWARD_ONLY));
bm |= GET_MESSAGE_BIT_TRANSMIT_TO;
k++;
}
- bf_size = GNUNET_CONTAINER_bloomfilter_get_size (pr->bf);
- 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;
- gm = (struct GetMessage *) lbuf;
- gm->header.type = htons (GNUNET_MESSAGE_TYPE_FS_GET);
- gm->header.size = htons (msize);
+ if (GNUNET_OK !=
+ GNUNET_BLOCK_group_serialize (pr->bg,
+ &bf_nonce,
+ &bf_data,
+ &bf_size))
+ {
+ bf_size = 0;
+ bf_data = NULL;
+ }
+ env = GNUNET_MQ_msg_extra (gm,
+ bf_size + k * sizeof (struct GNUNET_PeerIdentity),
+ GNUNET_MESSAGE_TYPE_FS_GET);
gm->type = htonl (pr->public_data.type);
if (do_route)
prio =
now = GNUNET_TIME_absolute_get ();
ttl = (int64_t) (pr->public_data.ttl.abs_value_us - now.abs_value_us);
gm->ttl = htonl (ttl / 1000LL / 1000LL);
- gm->filter_mutator = htonl (pr->mingle);
+ gm->filter_mutator = htonl (bf_nonce);
gm->hash_bitmap = htonl (bm);
gm->query = pr->public_data.query;
- ext = (struct GNUNET_HashCode *) & gm[1];
+ ext = (struct GNUNET_PeerIdentity *) &gm[1];
k = 0;
- if (!do_route)
+ if (! do_route)
GNUNET_PEER_resolve (pr->sender_pid,
- (struct GNUNET_PeerIdentity *) &ext[k++]);
+ &ext[k++]);
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,
- (char *) &ext[k],
- bf_size));
- memcpy (buf, gm, msize);
- return msize;
+ ext[k++] = *pr->public_data.target;
+ GNUNET_memcpy (&ext[k],
+ bf_data,
+ bf_size);
+ GNUNET_free_non_null (bf_data);
+ return env;
}
* @return #GNUNET_YES (we should continue to iterate)
*/
static int
-clean_request (void *cls, const struct GNUNET_HashCode *key, void *value)
+clean_request (void *cls,
+ const struct GNUNET_HashCode *key,
+ void *value)
{
struct GSF_PendingRequest *pr = value;
GSF_LocalLookupContinuation cont;
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
"Cleaning up pending request for `%s'.\n",
GNUNET_h2s (key));
- if (NULL != pr->mesh_request)
+ if (NULL != pr->cadet_request)
{
- pr->mesh_retry_count = MESH_RETRY_MAX;
- GSF_mesh_query_cancel (pr->mesh_request);
- pr->mesh_request = NULL;
+ pr->cadet_retry_count = CADET_RETRY_MAX;
+ GSF_cadet_query_cancel (pr->cadet_request);
+ pr->cadet_request = NULL;
}
if (NULL != (cont = pr->llc_cont))
{
pr->llc_cont = NULL;
- cont (pr->llc_cont_cls, pr, pr->local_result);
+ cont (pr->llc_cont_cls,
+ pr,
+ pr->local_result);
}
GSF_plan_notify_request_done_ (pr);
GNUNET_free_non_null (pr->replies_seen);
- if (NULL != pr->bf)
- {
- GNUNET_CONTAINER_bloomfilter_free (pr->bf);
- pr->bf = NULL;
- }
+ GNUNET_BLOCK_group_destroy (pr->bg);
+ pr->bg = NULL;
GNUNET_PEER_change_rc (pr->sender_pid, -1);
pr->sender_pid = 0;
GNUNET_PEER_change_rc (pr->origin_pid, -1);
GNUNET_DHT_get_stop (pr->gh);
pr->gh = NULL;
}
- if (GNUNET_SCHEDULER_NO_TASK != pr->warn_task)
+ if (NULL != pr->warn_task)
{
GNUNET_SCHEDULER_cancel (pr->warn_task);
- pr->warn_task = GNUNET_SCHEDULER_NO_TASK;
+ pr->warn_task = NULL;
}
GNUNET_assert (GNUNET_OK ==
GNUNET_CONTAINER_multihashmap_remove (pr_map,
&pr->public_data.query,
pr));
GNUNET_STATISTICS_update (GSF_stats,
- gettext_noop ("# Pending requests active"), -1,
+ gettext_noop ("# Pending requests active"),
+ -1,
GNUNET_NO);
GNUNET_free (pr);
return GNUNET_YES;
* @param full_cleanup fully purge the request
*/
void
-GSF_pending_request_cancel_ (struct GSF_PendingRequest *pr, int full_cleanup)
+GSF_pending_request_cancel_ (struct GSF_PendingRequest *pr,
+ int full_cleanup)
{
GSF_LocalLookupContinuation cont;
if (NULL == pr_map)
return; /* already cleaned up! */
- if (GNUNET_YES != full_cleanup)
+ if (GNUNET_NO == full_cleanup)
{
/* make request inactive (we're no longer interested in more results),
* but do NOT remove from our data-structures, we still need it there
* to prevent the request from looping */
pr->rh = NULL;
- if (NULL != pr->mesh_request)
+ if (NULL != pr->cadet_request)
{
- pr->mesh_retry_count = MESH_RETRY_MAX;
- GSF_mesh_query_cancel (pr->mesh_request);
- pr->mesh_request = NULL;
+ pr->cadet_retry_count = CADET_RETRY_MAX;
+ GSF_cadet_query_cancel (pr->cadet_request);
+ pr->cadet_request = NULL;
}
if (NULL != (cont = pr->llc_cont))
{
pr->llc_cont = NULL;
- cont (pr->llc_cont_cls, pr, pr->local_result);
+ cont (pr->llc_cont_cls,
+ pr,
+ pr->local_result);
}
GSF_plan_notify_request_done_ (pr);
if (NULL != pr->qe)
GNUNET_DHT_get_stop (pr->gh);
pr->gh = NULL;
}
- if (GNUNET_SCHEDULER_NO_TASK != pr->warn_task)
+ if (NULL != pr->warn_task)
{
GNUNET_SCHEDULER_cancel (pr->warn_task);
- pr->warn_task = GNUNET_SCHEDULER_NO_TASK;
+ pr->warn_task = NULL;
}
return;
}
GNUNET_assert (GNUNET_YES ==
- clean_request (NULL, &pr->public_data.query, pr));
+ clean_request (NULL,
+ &pr->public_data.query,
+ pr));
}
* Iterate over all pending requests.
*
* @param it function to call for each request
- * @param cls closure for it
+ * @param cls closure for @a it
*/
void
GSF_iterate_pending_requests_ (GSF_PendingRequestIterator it, void *cls)
*/
enum GNUNET_BLOCK_Type type;
+ /**
+ * Control flags for evaluation.
+ */
+ enum GNUNET_BLOCK_EvaluationOptions eo;
+
/**
* How much was this reply worth to us?
*/
/**
* We have received a reply; handle it!
*
- * @param cls response (struct ProcessReplyClosure)
+ * @param cls response (a `struct ProcessReplyClosure`)
* @param key our query
* @param value value in the hash map (info about the query)
* @return #GNUNET_YES (we should continue to iterate)
return GNUNET_YES;
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
"Matched result (type %u) for query `%s' with pending request\n",
- (unsigned int) prq->type, GNUNET_h2s (key));
+ (unsigned int) prq->type,
+ GNUNET_h2s (key));
GNUNET_STATISTICS_update (GSF_stats,
gettext_noop ("# replies received and matched"), 1,
GNUNET_NO);
prq->eval =
- GNUNET_BLOCK_evaluate (GSF_block_ctx, prq->type, key, &pr->bf, pr->mingle,
- NULL, 0, prq->data,
+ GNUNET_BLOCK_evaluate (GSF_block_ctx,
+ prq->type,
+ pr->bg,
+ prq->eo,
+ key,
+ NULL,
+ 0,
+ prq->data,
prq->size);
switch (prq->eval)
{
GNUNET_break (0);
return GNUNET_YES;
case GNUNET_BLOCK_EVALUATION_TYPE_NOT_SUPPORTED:
- GNUNET_log (GNUNET_ERROR_TYPE_ERROR, _("Unsupported block type %u\n"),
+ GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
+ _("Unsupported block type %u\n"),
prq->type);
return GNUNET_NO;
}
/* update bloomfilter */
- GNUNET_CRYPTO_hash (prq->data, prq->size, &chash);
- GSF_pending_request_update_ (pr, &chash, 1);
+ GNUNET_CRYPTO_hash (prq->data,
+ prq->size,
+ &chash);
+ GSF_pending_request_update_ (pr,
+ &chash,
+ 1);
if (NULL == prq->sender)
{
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
"Found result for query `%s' in local datastore\n",
GNUNET_h2s (key));
GNUNET_STATISTICS_update (GSF_stats,
- gettext_noop ("# results found locally"), 1,
+ gettext_noop ("# results found locally"),
+ 1,
GNUNET_NO);
}
else
prq->sender,
&last_transmission))
last_transmission.abs_value_us = GNUNET_TIME_UNIT_FOREVER_ABS.abs_value_us;
- pr->rh (pr->rh_cls, prq->eval, pr,
- prq->anonymity_level, prq->expiration,
- last_transmission, prq->type, prq->data, prq->size);
+ pr->rh (pr->rh_cls,
+ prq->eval,
+ pr,
+ prq->anonymity_level,
+ prq->expiration,
+ last_transmission,
+ prq->type,
+ prq->data,
+ prq->size);
return GNUNET_YES;
}
struct GNUNET_PeerIdentity origin;
/**
- * GNUNET_YES if we had a matching request for this block,
- * GNUNET_NO if not.
+ * #GNUNET_YES if we had a matching request for this block,
+ * #GNUNET_NO if not.
*/
int requested;
};
ppd->migration_delay);
mig_pause.rel_value_us = GNUNET_CRYPTO_random_u64 (GNUNET_CRYPTO_QUALITY_WEAK,
ppd->migration_delay.rel_value_us);
- ppd->migration_delay = GNUNET_TIME_relative_multiply (ppd->migration_delay, 2);
+ ppd->migration_delay = GNUNET_TIME_relative_saturating_multiply (ppd->migration_delay, 2);
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
"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));
+ GNUNET_STRINGS_relative_time_to_string (mig_pause,
+ GNUNET_YES));
+ GSF_block_peer_migration_ (cp,
+ GNUNET_TIME_relative_to_absolute (mig_pause));
}
}
GNUNET_free (pmc);
* @param data pointer to the result data
*/
static void
-handle_dht_reply (void *cls, struct GNUNET_TIME_Absolute exp,
- const struct GNUNET_HashCode * key,
+handle_dht_reply (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)
+ unsigned int put_path_length,
+ enum GNUNET_BLOCK_Type type,
+ size_t size,
+ const void *data)
{
struct GSF_PendingRequest *pr = cls;
struct ProcessReplyClosure prq;
prq.expiration);
prq.size = size;
prq.type = type;
+ prq.eo = GNUNET_BLOCK_EO_NONE;
process_reply (&prq, key, pr);
if ((GNUNET_YES == active_to_migration) &&
(GNUNET_NO == test_put_load_too_high (prq.priority)))
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
"Replicating result for query `%s' with priority %u\n",
GNUNET_h2s (key), prq.priority);
- pmc = GNUNET_malloc (sizeof (struct PutMigrationContext));
+ pmc = GNUNET_new (struct PutMigrationContext);
pmc->start = GNUNET_TIME_absolute_get ();
pmc->requested = GNUNET_YES;
if (NULL ==
1 /* anonymity */ ,
0 /* replication */ ,
exp, 1 + prq.priority, MAX_DATASTORE_QUEUE,
- GNUNET_CONSTANTS_SERVICE_TIMEOUT,
&put_migration_continuation, pmc))
{
- put_migration_continuation (pmc, GNUNET_SYSERR, GNUNET_TIME_UNIT_ZERO_ABS, NULL);
+ put_migration_continuation (pmc,
+ GNUNET_SYSERR,
+ GNUNET_TIME_UNIT_ZERO_ABS,
+ NULL);
}
}
}
{
GNUNET_assert (0 != pr->sender_pid);
GNUNET_PEER_resolve (pr->sender_pid, &pi);
- memcpy (&buf[xquery_size], &pi, sizeof (struct GNUNET_PeerIdentity));
+ GNUNET_memcpy (&buf[xquery_size], &pi, sizeof (struct GNUNET_PeerIdentity));
xquery_size += sizeof (struct GNUNET_PeerIdentity);
}
pr->gh =
/**
- * Function called with a reply from the mesh.
+ * Function called with a reply from the cadet.
*
* @param cls the pending request struct
* @param type type of the block, ANY on error
* @param data reply block data, NULL on error
*/
static void
-mesh_reply_proc (void *cls,
+cadet_reply_proc (void *cls,
enum GNUNET_BLOCK_Type type,
struct GNUNET_TIME_Absolute expiration,
size_t data_size,
struct ProcessReplyClosure prq;
struct GNUNET_HashCode query;
- pr->mesh_request = NULL;
+ pr->cadet_request = NULL;
if (GNUNET_BLOCK_TYPE_ANY == type)
{
GNUNET_break (NULL == data);
GNUNET_break (0 == data_size);
- pr->mesh_retry_count++;
- if (pr->mesh_retry_count >= MESH_RETRY_MAX)
- return; /* give up on mesh */
+ pr->cadet_retry_count++;
+ if (pr->cadet_retry_count >= CADET_RETRY_MAX)
+ return; /* give up on cadet */
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Error retrieiving block via mesh\n");
+ "Error retrieiving block via cadet\n");
/* retry -- without delay, as this is non-anonymous
- and mesh/mesh connect will take some time anyway */
- pr->mesh_request = GSF_mesh_query (pr->public_data.target,
- &pr->public_data.query,
- pr->public_data.type,
- &mesh_reply_proc,
- pr);
+ and cadet/cadet connect will take some time anyway */
+ pr->cadet_request = GSF_cadet_query (pr->public_data.target,
+ &pr->public_data.query,
+ pr->public_data.type,
+ &cadet_reply_proc,
+ pr);
return;
}
if (GNUNET_YES !=
return;
}
GNUNET_STATISTICS_update (GSF_stats,
- gettext_noop ("# Replies received from MESH"), 1,
+ gettext_noop ("# Replies received from CADET"), 1,
GNUNET_NO);
memset (&prq, 0, sizeof (prq));
prq.data = data;
prq.expiration);
prq.size = data_size;
prq.type = type;
+ prq.eo = GNUNET_BLOCK_EO_NONE;
process_reply (&prq, &query, pr);
}
/**
- * Consider downloading via mesh (if possible)
+ * Consider downloading via cadet (if possible)
*
* @param pr the pending request to process
*/
void
-GSF_mesh_lookup_ (struct GSF_PendingRequest *pr)
+GSF_cadet_lookup_ (struct GSF_PendingRequest *pr)
{
if (0 != pr->public_data.anonymity_level)
return;
if (0 == pr->public_data.target)
{
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Cannot do mesh-based download, target peer not known\n");
+ "Cannot do cadet-based download, target peer not known\n");
return;
}
- if (NULL != pr->mesh_request)
+ if (NULL != pr->cadet_request)
return;
- pr->mesh_request = GSF_mesh_query (pr->public_data.target,
- &pr->public_data.query,
- pr->public_data.type,
- &mesh_reply_proc,
- pr);
+ pr->cadet_request = GSF_cadet_query (pr->public_data.target,
+ &pr->public_data.query,
+ pr->public_data.type,
+ &cadet_reply_proc,
+ pr);
}
/**
* Task that issues a warning if the datastore lookup takes too long.
*
- * @param cls the 'struct GSF_PendingRequest'
- * @param tc task context
+ * @param cls the `struct GSF_PendingRequest`
*/
static void
-warn_delay_task (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
+warn_delay_task (void *cls)
{
struct GSF_PendingRequest *pr = cls;
- GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+ GNUNET_log (GNUNET_ERROR_TYPE_WARNING | GNUNET_ERROR_TYPE_BULK,
_("Datastore lookup already took %s!\n"),
- GNUNET_STRINGS_relative_time_to_string (GNUNET_TIME_absolute_get_duration (pr->qe_start), GNUNET_YES));
+ 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,
+ GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_MINUTES,
+ &warn_delay_task,
pr);
}
/**
* Task that issues a warning if the datastore lookup takes too long.
*
- * @param cls the 'struct GSF_PendingRequest'
- * @param tc task context
+ * @param cls the `struct GSF_PendingRequest`
*/
static void
-odc_warn_delay_task (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
+odc_warn_delay_task (void *cls)
{
struct GSF_PendingRequest *pr = cls;
* peer and if we are done either clean up (operation
* complete) or forward to other peers (more results possible).
*
- * @param cls our closure (struct PendingRequest)
+ * @param cls our closure (`struct GSF_PendingRequest *`)
* @param key key for the content
- * @param size number of bytes in data
+ * @param size number of bytes in @a data
* @param data content stored
* @param type type of the content
* @param priority priority of the content
* maybe 0 if no unique identifier is available
*/
static void
-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)
+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;
unsigned int old_rf;
GNUNET_SCHEDULER_cancel (pr->warn_task);
- pr->warn_task = GNUNET_SCHEDULER_NO_TASK;
+ pr->warn_task = NULL;
if (NULL != pr->qe)
{
pr->qe = NULL;
pr->public_data.options)) ? UINT_MAX :
GSF_datastore_queue_size
/* max queue size */ ,
- GNUNET_TIME_UNIT_FOREVER_REL,
&process_local_reply, pr);
if (NULL != pr->qe)
return; /* we're done */
{
GNUNET_break (0);
GNUNET_DATASTORE_remove (GSF_dsh, key, size, data, -1, -1,
- GNUNET_TIME_UNIT_FOREVER_REL, NULL, NULL);
+ NULL, NULL);
pr->qe_start = GNUNET_TIME_absolute_get ();
pr->warn_task =
GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_MINUTES,
pr->public_data.options)) ? UINT_MAX :
GSF_datastore_queue_size
/* max queue size */ ,
- GNUNET_TIME_UNIT_FOREVER_REL,
&process_local_reply, pr);
- if (pr->qe == NULL)
+ if (NULL == pr->qe)
{
GNUNET_STATISTICS_update (GSF_stats,
gettext_noop
prq.priority = priority;
prq.request_found = GNUNET_NO;
prq.anonymity_level = anonymity;
- if ((old_rf == 0) && (pr->public_data.results_found == 0))
+ if ((0 == old_rf) && (0 == pr->public_data.results_found))
GSF_update_datastore_delay_ (pr->public_data.start_time);
+ prq.eo = GNUNET_BLOCK_EO_LOCAL_SKIP_CRYPTO;
process_reply (&prq, key, pr);
pr->local_result = prq.eval;
if (prq.eval == GNUNET_BLOCK_EVALUATION_OK_LAST)
GNUNET_STATISTICS_update (GSF_stats,
gettext_noop
("# Datastore lookups concluded (found last result)"),
- 1, GNUNET_NO);
+ 1,
+ GNUNET_NO);
goto check_error_and_continue;
}
if ((0 == (GSF_PRO_PRIORITY_UNLIMITED & pr->public_data.options)) &&
((GNUNET_YES == GSF_test_get_load_too_high_ (0)) ||
(pr->public_data.results_found > 5 + 2 * pr->public_data.priority)))
{
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Load too high, done with request\n");
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Load too high, done with request\n");
GNUNET_STATISTICS_update (GSF_stats,
- gettext_noop
- ("# Datastore lookups concluded (load too high)"),
- 1, GNUNET_NO);
+ gettext_noop ("# Datastore lookups concluded (load too high)"),
+ 1,
+ GNUNET_NO);
goto check_error_and_continue;
}
pr->qe_start = GNUNET_TIME_absolute_get ();
pr->warn_task =
- GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_MINUTES, &warn_delay_task,
+ GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_MINUTES,
+ &warn_delay_task,
pr);
pr->qe =
GNUNET_DATASTORE_get_key (GSF_dsh, pr->local_result_offset++,
public_data.options)) ? UINT_MAX :
GSF_datastore_queue_size
/* max queue size */ ,
- GNUNET_TIME_UNIT_FOREVER_REL,
&process_local_reply, pr);
/* check if we successfully queued another datastore request;
* if so, return, otherwise call our continuation (if we have
check_error_and_continue:
if (NULL != pr->qe)
return;
- if (GNUNET_SCHEDULER_NO_TASK != pr->warn_task)
+ if (NULL != pr->warn_task)
{
GNUNET_SCHEDULER_cancel (pr->warn_task);
- pr->warn_task = GNUNET_SCHEDULER_NO_TASK;
+ pr->warn_task = NULL;
}
if (NULL == (cont = pr->llc_cont))
return; /* no continuation */
pr->llc_cont = NULL;
+ if (0 != (GSF_PRO_LOCAL_ONLY & pr->public_data.options))
+ {
+ if (GNUNET_BLOCK_EVALUATION_OK_LAST != pr->local_result)
+ {
+ /* Signal that we are done and that there won't be any
+ additional results to allow client to clean up state. */
+ pr->rh (pr->rh_cls,
+ GNUNET_BLOCK_EVALUATION_OK_LAST,
+ pr,
+ UINT32_MAX,
+ GNUNET_TIME_UNIT_ZERO_ABS,
+ GNUNET_TIME_UNIT_FOREVER_ABS,
+ GNUNET_BLOCK_TYPE_ANY,
+ NULL, 0);
+ }
+ /* Finally, call our continuation to signal that we are
+ done with local processing of this request; i.e. to
+ start reading again from the client. */
+ cont (pr->llc_cont_cls, NULL, GNUNET_BLOCK_EVALUATION_OK_LAST);
+ return;
+ }
+
cont (pr->llc_cont_cls, pr, pr->local_result);
}
*
* @param pr request
* @param target
- * @return GNUNET_YES if this request could be forwarded to the given peer
+ * @return #GNUNET_YES if this request could be forwarded to the given peer
*/
int
GSF_pending_request_test_target_ (struct GSF_PendingRequest *pr,
*
* @param pr the pending request to process
* @param cont function to call at the end
- * @param cont_cls closure for cont
+ * @param cont_cls closure for @a cont
*/
void
GSF_local_lookup_ (struct GSF_PendingRequest *pr,
- GSF_LocalLookupContinuation cont, void *cont_cls)
+ GSF_LocalLookupContinuation cont,
+ void *cont_cls)
{
GNUNET_assert (NULL == pr->gh);
- GNUNET_assert (NULL == pr->mesh_request);
+ GNUNET_assert (NULL == pr->cadet_request);
GNUNET_assert (NULL == pr->llc_cont);
pr->llc_cont = cont;
pr->llc_cont_cls = cont_cls;
pr->qe_start = GNUNET_TIME_absolute_get ();
pr->warn_task =
- GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_MINUTES, &warn_delay_task,
+ GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_MINUTES,
+ &warn_delay_task,
pr);
#if INSANE_STATISTICS
GNUNET_STATISTICS_update (GSF_stats,
public_data.options)) ? UINT_MAX :
GSF_datastore_queue_size
/* max queue size */ ,
- GNUNET_TIME_UNIT_FOREVER_REL,
&process_local_reply, pr);
if (NULL != pr->qe)
return;
("# Datastore lookups concluded (error queueing)"),
1, GNUNET_NO);
GNUNET_SCHEDULER_cancel (pr->warn_task);
- pr->warn_task = GNUNET_SCHEDULER_NO_TASK;
+ pr->warn_task = NULL;
pr->llc_cont = NULL;
if (NULL != cont)
cont (cont_cls, pr, pr->local_result);
* this content and possibly passes it on (to local clients or other
* peers). Does NOT perform migration (content caching at this peer).
*
- * @param cp the other peer involved (sender or receiver, NULL
- * for loopback messages where we are both sender and receiver)
- * @param message the actual message
- * @return GNUNET_OK if the message was well-formed,
- * GNUNET_SYSERR if the message was malformed (close connection,
- * do not cache under any circumstances)
+ * @param cls the other peer involved
+ * @param put the actual message
*/
-int
-GSF_handle_p2p_content_ (struct GSF_ConnectedPeer *cp,
- const struct GNUNET_MessageHeader *message)
+void
+handle_p2p_put (void *cls,
+ const struct PutMessage *put)
{
- const struct PutMessage *put;
+ struct GSF_ConnectedPeer *cp = cls;
uint16_t msize;
size_t dsize;
enum GNUNET_BLOCK_Type type;
double putl;
struct PutMigrationContext *pmc;
- msize = ntohs (message->size);
- if (msize < sizeof (struct PutMessage))
- {
- GNUNET_break_op (0);
- return GNUNET_SYSERR;
- }
- put = (const struct PutMessage *) message;
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Received P2P PUT from %s\n",
+ GNUNET_i2s (GSF_get_peer_performance_data_ (cp)->peer));
+ GSF_cover_content_count++;
+ msize = ntohs (put->header.size);
dsize = msize - sizeof (struct PutMessage);
type = ntohl (put->type);
expiration = GNUNET_TIME_absolute_ntoh (put->expiration);
/* do not allow migrated content to live longer than 1 year */
expiration = GNUNET_TIME_absolute_min (GNUNET_TIME_relative_to_absolute (GNUNET_TIME_UNIT_YEARS),
expiration);
- if (type == GNUNET_BLOCK_TYPE_FS_ONDEMAND)
- return GNUNET_SYSERR;
if (GNUNET_OK !=
- GNUNET_BLOCK_get_key (GSF_block_ctx, type, &put[1], dsize, &query))
+ GNUNET_BLOCK_get_key (GSF_block_ctx,
+ type,
+ &put[1],
+ dsize,
+ &query))
{
GNUNET_break_op (0);
- return GNUNET_SYSERR;
+ return;
}
GNUNET_STATISTICS_update (GSF_stats,
- gettext_noop ("# GAP PUT messages received"), 1,
+ gettext_noop ("# GAP PUT messages received"),
+ 1,
GNUNET_NO);
/* now, lookup 'query' */
prq.data = (const void *) &put[1];
prq.priority = 0;
prq.anonymity_level = UINT32_MAX;
prq.request_found = GNUNET_NO;
- GNUNET_CONTAINER_multihashmap_get_multiple (pr_map, &query, &process_reply,
+ prq.eo = GNUNET_BLOCK_EO_NONE;
+ GNUNET_CONTAINER_multihashmap_get_multiple (pr_map,
+ &query,
+ &process_reply,
&prq);
if (NULL != cp)
{
{
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
"Replicating result for query `%s' with priority %u\n",
- GNUNET_h2s (&query), prq.priority);
- pmc = GNUNET_malloc (sizeof (struct PutMigrationContext));
+ GNUNET_h2s (&query),
+ prq.priority);
+ pmc = GNUNET_new (struct PutMigrationContext);
pmc->start = GNUNET_TIME_absolute_get ();
pmc->requested = prq.request_found;
GNUNET_assert (0 != GSF_get_peer_performance_data_ (cp)->pid);
GNUNET_PEER_resolve (GSF_get_peer_performance_data_ (cp)->pid,
&pmc->origin);
if (NULL ==
- GNUNET_DATASTORE_put (GSF_dsh, 0, &query, dsize, &put[1], type,
- prq.priority, 1 /* anonymity */ ,
+ GNUNET_DATASTORE_put (GSF_dsh,
+ 0,
+ &query,
+ dsize,
+ &put[1],
+ type,
+ prq.priority,
+ 1 /* anonymity */ ,
0 /* replication */ ,
- expiration, 1 + prq.priority, MAX_DATASTORE_QUEUE,
- GNUNET_CONSTANTS_SERVICE_TIMEOUT,
- &put_migration_continuation, pmc))
+ expiration, 1 + prq.priority,
+ MAX_DATASTORE_QUEUE,
+ &put_migration_continuation,
+ pmc))
{
- put_migration_continuation (pmc, GNUNET_SYSERR, GNUNET_TIME_UNIT_ZERO_ABS, NULL);
+ put_migration_continuation (pmc,
+ GNUNET_SYSERR,
+ GNUNET_TIME_UNIT_ZERO_ABS,
+ NULL);
}
}
else if (NULL != cp)
{
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
"Choosing not to keep content `%s' (%d/%d)\n",
- GNUNET_h2s (&query), active_to_migration,
+ GNUNET_h2s (&query),
+ active_to_migration,
test_put_load_too_high (prq.priority));
}
putl = GNUNET_LOAD_get_load (datastore_put_load);
putl,
active_to_migration,
(GNUNET_NO == prq.request_found));
- GSF_block_peer_migration_ (cp, GNUNET_TIME_relative_to_absolute (block_time));
+ GSF_block_peer_migration_ (cp,
+ GNUNET_TIME_relative_to_absolute (block_time));
}
- return GNUNET_OK;
+}
+
+
+/**
+ * Check if the given request is still active.
+ *
+ * @param pr pending request
+ * @return #GNUNET_YES if the request is still active
+ */
+int
+GSF_pending_request_test_active_ (struct GSF_PendingRequest *pr)
+{
+ return (NULL != pr->rh) ? GNUNET_YES : GNUNET_NO;
}
void
GSF_pending_request_done_ ()
{
- GNUNET_CONTAINER_multihashmap_iterate (pr_map, &clean_request, NULL);
+ GNUNET_CONTAINER_multihashmap_iterate (pr_map,
+ &clean_request,
+ NULL);
GNUNET_CONTAINER_multihashmap_destroy (pr_map);
pr_map = NULL;
GNUNET_CONTAINER_heap_destroy (requests_by_expiration_heap);
projects / oweals / gnunet.git / blobdiff