This file is part of GNUnet.
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
- by the Free Software Foundation; either version 3, or (at your
- option) any later version.
+ GNUnet is free software: you can redistribute it and/or modify it
+ under the terms of the GNU Affero General Public License as published
+ by the Free Software Foundation, either version 3 of the License,
+ or (at your option) any later version.
GNUnet is distributed in the hope that it will be useful, but
WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
- General Public License for more details.
+ Affero General Public License for more details.
+
+ You should have received a copy of the GNU Affero General Public License
+ along with this program. If not, see <http://www.gnu.org/licenses/>.
- 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., 51 Franklin Street, Fifth Floor,
- Boston, MA 02110-1301, USA.
+ SPDX-License-Identifier: AGPL3.0-or-later
*/
/**
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_SCHEDULER_Task * warn_task;
/**
- * Current offset for querying our local datastore for results.
- * Starts at a random value, incremented until we get the same
- * UID again (detected using 'first_uid'), which is then used
- * to termiante the iteration.
+ * Do we have a first UID yet?
+ */
+ bool have_first_uid;
+
+ /**
+ * Have we seen a NULL result yet?
*/
- uint64_t local_result_offset;
+ bool seen_null;
/**
* Unique ID of the first result from the local datastore;
- * used to detect wrap-around of the offset.
+ * used to terminate the loop.
*/
uint64_t first_uid;
+ /**
+ * Result count.
+ */
+ size_t result_count;
+
/**
* How often have we retried this request via 'cadet'?
* (used to bound overall retries).
*/
unsigned int replies_seen_size;
- /**
- * Mingle value we currently use for the bf.
- */
- uint32_t mingle;
-
- /**
- * Do we have a first UID yet?
- */
- unsigned int have_first_uid;
-
};
* 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));
}
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 (NULL != target)
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);
+ 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));
+ 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,
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);
- GNUNET_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)
struct GNUNET_TIME_Absolute now;
int64_t ttl;
int do_route;
+ void *bf_data;
+ uint32_t bf_nonce;
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
"Building request message for `%s' of type %d\n",
bm |= GET_MESSAGE_BIT_TRANSMIT_TO;
k++;
}
- bf_size = GNUNET_CONTAINER_bloomfilter_get_size (pr->bf);
+ 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);
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_PeerIdentity *) &gm[1];
&ext[k++]);
if (NULL != pr->public_data.target)
ext[k++] = *pr->public_data.target;
- if (NULL != pr->bf)
- GNUNET_assert (GNUNET_SYSERR !=
- GNUNET_CONTAINER_bloomfilter_get_raw_data (pr->bf,
- (char *) &ext[k],
- bf_size));
+ GNUNET_memcpy (&ext[k],
+ bf_data,
+ bf_size);
+ GNUNET_free_non_null (bf_data);
return env;
}
}
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);
prq->eval =
GNUNET_BLOCK_evaluate (GSF_block_ctx,
prq->type,
+ pr->bg,
prq->eo,
key,
- &pr->bf,
- pr->mingle,
NULL,
0,
prq->data,
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,
}
+/* Call our continuation (if we have any) */
+static void
+call_continuation (struct GSF_PendingRequest *pr)
+{
+ GSF_LocalLookupContinuation cont = pr->llc_cont;
+
+ GNUNET_assert (NULL == pr->qe);
+ if (NULL != pr->warn_task)
+ {
+ GNUNET_SCHEDULER_cancel (pr->warn_task);
+ pr->warn_task = NULL;
+ }
+ if (NULL == 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);
+}
+
+
+/* Update stats and call continuation */
+static void
+no_more_local_results (struct GSF_PendingRequest *pr)
+{
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG | GNUNET_ERROR_TYPE_BULK,
+ "No further local responses available.\n");
+#if INSANE_STATISTICS
+ if ( (GNUNET_BLOCK_TYPE_FS_DBLOCK == pr->public_data.type) ||
+ (GNUNET_BLOCK_TYPE_FS_IBLOCK == pr->public_data.type) )
+ GNUNET_STATISTICS_update (GSF_stats,
+ gettext_noop ("# requested DBLOCK or IBLOCK not found"),
+ 1,
+ GNUNET_NO);
+#endif
+ call_continuation (pr);
+}
+
+
+/* forward declaration */
+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,
+ uint32_t replication,
+ struct GNUNET_TIME_Absolute expiration,
+ uint64_t uid);
+
+
+/* Start a local query */
+static void
+start_local_query (struct GSF_PendingRequest *pr,
+ uint64_t next_uid,
+ bool random)
+{
+ pr->qe_start = GNUNET_TIME_absolute_get ();
+ pr->warn_task =
+ GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_MINUTES,
+ &warn_delay_task,
+ pr);
+ pr->qe =
+ GNUNET_DATASTORE_get_key (GSF_dsh,
+ next_uid,
+ random,
+ &pr->public_data.query,
+ pr->public_data.type ==
+ GNUNET_BLOCK_TYPE_FS_DBLOCK ?
+ GNUNET_BLOCK_TYPE_ANY : pr->public_data.type,
+ (0 !=
+ (GSF_PRO_PRIORITY_UNLIMITED & pr->
+ public_data.options)) ? UINT_MAX : 1
+ /* queue priority */ ,
+ (0 !=
+ (GSF_PRO_PRIORITY_UNLIMITED & pr->
+ public_data.options)) ? UINT_MAX :
+ GSF_datastore_queue_size
+ /* max queue size */ ,
+ &process_local_reply, pr);
+ if (NULL != pr->qe)
+ return;
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "ERROR Requesting `%s' of type %d with next_uid %llu from datastore.\n",
+ GNUNET_h2s (&pr->public_data.query),
+ pr->public_data.type,
+ (unsigned long long) next_uid);
+ GNUNET_STATISTICS_update (GSF_stats,
+ gettext_noop ("# Datastore lookups concluded (error queueing)"),
+ 1,
+ GNUNET_NO);
+ call_continuation (pr);
+}
+
+
/**
* We're processing (local) results for a search request
* from another peer. Pass applicable results to the
* @param type type of the content
* @param priority priority of the content
* @param anonymity anonymity-level for the content
+ * @param replication replication-level for the content
* @param expiration expiration time for the content
* @param uid unique identifier for the datum;
* maybe 0 if no unique identifier is available
enum GNUNET_BLOCK_Type type,
uint32_t priority,
uint32_t anonymity,
+ uint32_t replication,
struct GNUNET_TIME_Absolute expiration,
uint64_t uid)
{
struct GSF_PendingRequest *pr = cls;
- GSF_LocalLookupContinuation cont;
struct ProcessReplyClosure prq;
struct GNUNET_HashCode query;
unsigned int old_rf;
GNUNET_SCHEDULER_cancel (pr->warn_task);
pr->warn_task = NULL;
- if (NULL != pr->qe)
+ if (NULL == pr->qe)
+ goto called_from_on_demand;
+ pr->qe = NULL;
+ if ( (NULL == key) &&
+ pr->seen_null &&
+ !pr->have_first_uid) /* We have hit the end for the 2nd time with no results */
{
- pr->qe = NULL;
- if (NULL == key)
- {
+ /* No results */
#if INSANE_STATISTICS
- GNUNET_STATISTICS_update (GSF_stats,
- gettext_noop
- ("# Datastore lookups concluded (no results)"),
- 1, GNUNET_NO);
+ GNUNET_STATISTICS_update (GSF_stats,
+ gettext_noop
+ ("# Datastore lookups concluded (no results)"),
+ 1, GNUNET_NO);
#endif
- }
- if (GNUNET_NO == pr->have_first_uid)
- {
- pr->first_uid = uid;
- pr->have_first_uid = 1;
- }
- else
- {
- if ((uid == pr->first_uid) && (key != NULL))
- {
- GNUNET_STATISTICS_update (GSF_stats,
- gettext_noop
- ("# Datastore lookups concluded (seen all)"),
- 1, GNUNET_NO);
- key = NULL; /* all replies seen! */
- }
- pr->have_first_uid++;
- if ((pr->have_first_uid > MAX_RESULTS) && (key != NULL))
- {
- GNUNET_STATISTICS_update (GSF_stats,
- gettext_noop
- ("# Datastore lookups aborted (more than MAX_RESULTS)"),
- 1, GNUNET_NO);
- key = NULL; /* all replies seen! */
- }
- }
+ no_more_local_results (pr);
+ return;
+ }
+ if ( ( (NULL == key) &&
+ pr->seen_null ) || /* We have hit the end for the 2nd time OR */
+ ( pr->seen_null &&
+ pr->have_first_uid &&
+ (uid >= pr->first_uid) ) ) /* We have hit the end and past first UID */
+ {
+ /* Seen all results */
+ GNUNET_STATISTICS_update (GSF_stats,
+ gettext_noop
+ ("# Datastore lookups concluded (seen all)"),
+ 1, GNUNET_NO);
+ no_more_local_results (pr);
+ return;
}
if (NULL == key)
{
- 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_assert (!pr->seen_null);
+ pr->seen_null = true;
+ start_local_query (pr,
+ 0 /* next_uid */,
+ false /* random */);
+ return;
+ }
+ if (!pr->have_first_uid)
+ {
+ pr->first_uid = uid;
+ pr->have_first_uid = true;
+ }
+ pr->result_count++;
+ if (pr->result_count > MAX_RESULTS)
+ {
+ GNUNET_STATISTICS_update (GSF_stats,
+ gettext_noop
+ ("# Datastore lookups aborted (more than MAX_RESULTS)"),
+ 1, GNUNET_NO);
+ no_more_local_results (pr);
+ return;
}
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
"Received reply for `%s' of type %d with UID %llu from datastore.\n",
GNUNET_h2s (key), type, (unsigned long long) uid);
- if (type == GNUNET_BLOCK_TYPE_FS_ONDEMAND)
+ if (GNUNET_BLOCK_TYPE_FS_ONDEMAND == type)
{
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
"Found ONDEMAND block, performing on-demand encoding\n");
GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_MINUTES,
&odc_warn_delay_task, pr);
if (GNUNET_OK ==
- GNUNET_FS_handle_on_demand_block (key, size, data, type, priority,
- anonymity, expiration, uid,
- &process_local_reply, pr))
+ GNUNET_FS_handle_on_demand_block (key,
+ size,
+ data,
+ type,
+ priority,
+ anonymity,
+ replication,
+ expiration,
+ uid,
+ &process_local_reply,
+ pr))
{
GNUNET_STATISTICS_update (GSF_stats,
gettext_noop
gettext_noop ("# on-demand lookups failed"), 1,
GNUNET_NO);
GNUNET_SCHEDULER_cancel (pr->warn_task);
- pr->warn_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 - 1,
- &pr->public_data.query,
- pr->public_data.type ==
- GNUNET_BLOCK_TYPE_FS_DBLOCK ?
- GNUNET_BLOCK_TYPE_ANY : pr->public_data.type,
- (0 !=
- (GSF_PRO_PRIORITY_UNLIMITED &
- pr->public_data.options)) ? UINT_MAX : 1
- /* queue priority */ ,
- (0 !=
- (GSF_PRO_PRIORITY_UNLIMITED &
- pr->public_data.options)) ? UINT_MAX :
- GSF_datastore_queue_size
- /* max queue size */ ,
- &process_local_reply, pr);
- if (NULL != pr->qe)
- return; /* we're done */
- GNUNET_STATISTICS_update (GSF_stats,
- gettext_noop
- ("# Datastore lookups concluded (error queueing)"),
- 1, GNUNET_NO);
- goto check_error_and_continue;
+ start_local_query (pr,
+ uid + 1 /* next_uid */,
+ false /* random */);
+ return;
}
+called_from_on_demand:
old_rf = pr->public_data.results_found;
memset (&prq, 0, sizeof (prq));
prq.data = data;
GNUNET_BLOCK_get_key (GSF_block_ctx, type, data, size, &query))
{
GNUNET_break (0);
- GNUNET_DATASTORE_remove (GSF_dsh, key, size, data, -1, -1,
+ GNUNET_DATASTORE_remove (GSF_dsh, key, size, data, UINT_MAX, UINT_MAX,
NULL, NULL);
- pr->qe_start = GNUNET_TIME_absolute_get ();
- pr->warn_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 - 1,
- &pr->public_data.query,
- pr->public_data.type ==
- GNUNET_BLOCK_TYPE_FS_DBLOCK ?
- GNUNET_BLOCK_TYPE_ANY : pr->public_data.type,
- (0 !=
- (GSF_PRO_PRIORITY_UNLIMITED &
- pr->public_data.options)) ? UINT_MAX : 1
- /* queue priority */ ,
- (0 !=
- (GSF_PRO_PRIORITY_UNLIMITED &
- pr->public_data.options)) ? UINT_MAX :
- GSF_datastore_queue_size
- /* max queue size */ ,
- &process_local_reply, pr);
- if (NULL == pr->qe)
- {
- GNUNET_STATISTICS_update (GSF_stats,
- gettext_noop
- ("# Datastore lookups concluded (error queueing)"),
- 1, GNUNET_NO);
- goto check_error_and_continue;
- }
+ start_local_query (pr,
+ uid + 1 /* next_uid */,
+ false /* random */);
return;
}
prq.type = type;
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)
+ if (GNUNET_BLOCK_EVALUATION_OK_LAST == prq.eval)
{
GNUNET_STATISTICS_update (GSF_stats,
gettext_noop
("# Datastore lookups concluded (found last result)"),
1,
GNUNET_NO);
- goto check_error_and_continue;
+ call_continuation (pr);
+ return;
}
if ((0 == (GSF_PRO_PRIORITY_UNLIMITED & pr->public_data.options)) &&
((GNUNET_YES == GSF_test_get_load_too_high_ (0)) ||
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,
- pr);
- pr->qe =
- GNUNET_DATASTORE_get_key (GSF_dsh, pr->local_result_offset++,
- &pr->public_data.query,
- pr->public_data.type ==
- GNUNET_BLOCK_TYPE_FS_DBLOCK ?
- GNUNET_BLOCK_TYPE_ANY : pr->public_data.type,
- (0 !=
- (GSF_PRO_PRIORITY_UNLIMITED & pr->
- public_data.options)) ? UINT_MAX : 1
- /* queue priority */ ,
- (0 !=
- (GSF_PRO_PRIORITY_UNLIMITED & pr->
- public_data.options)) ? UINT_MAX :
- GSF_datastore_queue_size
- /* max queue size */ ,
- &process_local_reply, pr);
- /* check if we successfully queued another datastore request;
- * if so, return, otherwise call our continuation (if we have
- * any) */
-check_error_and_continue:
- if (NULL != pr->qe)
- return;
- if (NULL != pr->warn_task)
- {
- GNUNET_SCHEDULER_cancel (pr->warn_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);
+ call_continuation (pr);
return;
}
-
- cont (pr->llc_cont_cls, pr, pr->local_result);
+ start_local_query (pr,
+ uid + 1 /* next_uid */,
+ false /* random */);
}
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,
- 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,
- pr->public_data.type ==
- GNUNET_BLOCK_TYPE_FS_DBLOCK ?
- GNUNET_BLOCK_TYPE_ANY : pr->public_data.type,
- (0 !=
- (GSF_PRO_PRIORITY_UNLIMITED & pr->
- public_data.options)) ? UINT_MAX : 1
- /* queue priority */ ,
- (0 !=
- (GSF_PRO_PRIORITY_UNLIMITED & pr->
- public_data.options)) ? UINT_MAX :
- GSF_datastore_queue_size
- /* max queue size */ ,
- &process_local_reply, pr);
- if (NULL != pr->qe)
- return;
- GNUNET_STATISTICS_update (GSF_stats,
- gettext_noop
- ("# Datastore lookups concluded (error queueing)"),
- 1, GNUNET_NO);
- GNUNET_SCHEDULER_cancel (pr->warn_task);
- pr->warn_task = NULL;
- pr->llc_cont = NULL;
- if (NULL != cont)
- cont (cont_cls, pr, pr->local_result);
+ start_local_query(pr,
+ 0 /* next_uid */,
+ true /* random */);
}
projects / oweals / gnunet.git / blobdiff