gnunet-service-fs_pr.c File Reference

API to handle pending requests. More...

#include "platform.h"
#include "gnunet_util_lib.h"
#include "gnunet_load_lib.h"
#include "gnunet-service-fs.h"
#include "gnunet-service-fs_cp.h"
#include "gnunet-service-fs_indexing.h"
#include "gnunet-service-fs_pe.h"
#include "gnunet-service-fs_pr.h"
#include "gnunet-service-fs_cadet.h"

Include dependency graph for gnunet-service-fs_pr.c:

Go to the source code of this file.

Data Structures
struct	GSF_PendingRequest
	An active request. More...
struct	ProcessReplyClosure
	Closure for process_reply() function. More...
struct	PutMigrationContext
	Context for put_migration_continuation(). More...

Macros
#define	DHT_GET_REPLICATION   5
	Desired replication level for GETs. More...
#define	MAX_DATASTORE_QUEUE   (16 * MAX_QUEUE_PER_PEER)
	Maximum size of the datastore queue for P2P operations. More...
#define	CONTENT_BANDWIDTH_VALUE   800
	Bandwidth value of a 0-priority content (must be fairly high compared to query since content is typically significantly larger – and more valuable since it can take many queries to get one piece of content). More...
#define	MAX_RESULTS   (100 * 1024)
	Hard limit on the number of results we may get from the datastore per query. More...
#define	INSANE_STATISTICS   GNUNET_NO
	Collect an instance number of statistics? May cause excessive IPC. More...
#define	CADET_RETRY_MAX   3
	If obtaining a block via cadet fails, how often do we retry it before giving up for good (and sticking to non-anonymous transfer)? More...

Functions
static void	refresh_bloomfilter (enum GNUNET_BLOCK_Type type, struct GSF_PendingRequest *pr)
	Recalculate our bloom filter for filtering replies. More...
struct GSF_PendingRequest *	GSF_pending_request_create_ (enum GSF_PendingRequestOptions options, 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 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)
	Create a new pending request. More...
struct GSF_PendingRequestData *	GSF_pending_request_get_data_ (struct GSF_PendingRequest *pr)
	Obtain the public data associated with a pending request. More...
int	GSF_pending_request_is_compatible_ (struct GSF_PendingRequest *pra, struct GSF_PendingRequest *prb)
	Test if two pending requests are compatible (would generate the same query modulo filters and should thus be processed jointly). More...
void	GSF_pending_request_update_ (struct GSF_PendingRequest *pr, const struct GNUNET_HashCode *replies_seen, unsigned int replies_seen_count)
	Update a given pending request with additional replies that have been seen. More...
struct GNUNET_MQ_Envelope *	GSF_pending_request_get_message_ (struct GSF_PendingRequest *pr)
	Generate the message corresponding to the given pending request for transmission to other peers. More...
static int	clean_request (void *cls, const struct GNUNET_HashCode *key, void *value)
	Iterator to free pending requests. More...
void	GSF_pending_request_cancel_ (struct GSF_PendingRequest *pr, int full_cleanup)
	Explicitly cancel a pending request. More...
void	GSF_iterate_pending_requests_ (GSF_PendingRequestIterator it, void *cls)
	Iterate over all pending requests. More...
static void	update_request_performance_data (struct ProcessReplyClosure *prq, struct GSF_PendingRequest *pr)
	Update the performance data for the sender (if any) since the sender successfully answered one of our queries. More...
static enum GNUNET_GenericReturnValue	process_reply (void *cls, const struct GNUNET_HashCode *key, void *value)
	We have received a reply; handle it! More...
static void	put_migration_continuation (void *cls, int success, struct GNUNET_TIME_Absolute min_expiration, const char *msg)
	Continuation called to notify client about result of the operation. More...
static int	test_put_load_too_high (uint32_t priority)
	Test if the DATABASE (PUT) load on this peer is too high to even consider processing the query at all. More...
static void	handle_dht_reply (void *cls, struct GNUNET_TIME_Absolute exp, const struct GNUNET_HashCode *key, const struct GNUNET_PeerIdentity *trunc_peer, const struct GNUNET_DHT_PathElement *get_path, unsigned int get_path_length, const struct GNUNET_DHT_PathElement *put_path, unsigned int put_path_length, enum GNUNET_BLOCK_Type type, size_t size, const void *data)
	Iterator called on each result obtained for a DHT operation that expects a reply. More...
void	GSF_dht_lookup_ (struct GSF_PendingRequest *pr)
	Consider looking up the data in the DHT (anonymity-level permitting). More...
static void	cadet_reply_proc (void *cls, enum GNUNET_BLOCK_Type type, struct GNUNET_TIME_Absolute expiration, size_t data_size, const void *data)
	Function called with a reply from the cadet. More...
void	GSF_cadet_lookup_ (struct GSF_PendingRequest *pr)
	Consider downloading via cadet (if possible) More...
static void	warn_delay_task (void *cls)
	Task that issues a warning if the datastore lookup takes too long. More...
static void	odc_warn_delay_task (void *cls)
	Task that issues a warning if the datastore lookup takes too long. More...
static void	call_continuation (struct GSF_PendingRequest *pr)
static void	no_more_local_results (struct GSF_PendingRequest *pr)
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)
	We're processing (local) results for a search request from another peer. More...
static void	start_local_query (struct GSF_PendingRequest *pr, uint64_t next_uid, bool random)
int	GSF_pending_request_test_target_ (struct GSF_PendingRequest *pr, const struct GNUNET_PeerIdentity *target)
	Is the given target a legitimate peer for forwarding the given request? More...
void	GSF_local_lookup_ (struct GSF_PendingRequest *pr, GSF_LocalLookupContinuation cont, void *cont_cls)
	Look up the request in the local datastore. More...
void	handle_p2p_put (void *cls, const struct PutMessage *put)
	Handle P2P "CONTENT" message. More...
int	GSF_pending_request_test_active_ (struct GSF_PendingRequest *pr)
	Check if the given request is still active. More...
void	GSF_pending_request_init_ ()
	Setup the subsystem. More...
void	GSF_pending_request_done_ ()
	Shutdown the subsystem. More...

Variables
static struct GNUNET_CONTAINER_MultiHashMap *	pr_map
	All pending requests, ordered by the query. More...
static struct GNUNET_LOAD_Value *	datastore_put_load
	Datastore 'PUT' load tracking. More...
static int	active_to_migration
	Are we allowed to migrate content to this peer. More...
static struct GNUNET_CONTAINER_Heap *	requests_by_expiration_heap
	Heap with the request that will expire next at the top. More...
static unsigned long long	max_pending_requests = (32 * 1024)
	Maximum number of requests (from other peers, overall) that we're willing to have pending at any given point in time. More...

Detailed Description

API to handle pending requests.

Author

Christian Grothoff

Definition in file gnunet-service-fs_pr.c.

Macro Definition Documentation

DHT_GET_REPLICATION

#define DHT_GET_REPLICATION   5

Desired replication level for GETs.

Definition at line 40 of file gnunet-service-fs_pr.c.

MAX_DATASTORE_QUEUE

#define MAX_DATASTORE_QUEUE   (16 * MAX_QUEUE_PER_PEER)

Maximum size of the datastore queue for P2P operations.

Needs to be large enough to queue MAX_QUEUE_PER_PEER operations for roughly the number of active (connected) peers.

Definition at line 47 of file gnunet-service-fs_pr.c.

CONTENT_BANDWIDTH_VALUE

#define CONTENT_BANDWIDTH_VALUE   800

Bandwidth value of a 0-priority content (must be fairly high compared to query since content is typically significantly larger – and more valuable since it can take many queries to get one piece of content).

Definition at line 55 of file gnunet-service-fs_pr.c.

MAX_RESULTS

#define MAX_RESULTS   (100 * 1024)

Hard limit on the number of results we may get from the datastore per query.

Definition at line 60 of file gnunet-service-fs_pr.c.

INSANE_STATISTICS

#define INSANE_STATISTICS   GNUNET_NO

Collect an instance number of statistics? May cause excessive IPC.

Definition at line 65 of file gnunet-service-fs_pr.c.

CADET_RETRY_MAX

#define CADET_RETRY_MAX   3

If obtaining a block via cadet fails, how often do we retry it before giving up for good (and sticking to non-anonymous transfer)?

Definition at line 71 of file gnunet-service-fs_pr.c.

Function Documentation

refresh_bloomfilter()

static void refresh_bloomfilter ( enum GNUNET_BLOCK_Type  type, struct GSF_PendingRequest *  pr  )

static

Recalculate our bloom filter for filtering replies.

This function will create a new bloom filter from scratch, so it should only be called if we have no bloomfilter at all (and hence can create a fresh one of minimal size without problems) OR if our peer is the initiator (in which case we may resize to larger than minimum size).

Parameters

type	type of the request
pr	request for which the BF is to be recomputed

Definition at line 250 of file gnunet-service-fs_pr.c.

{
  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,
                               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));
}

References GSF_PendingRequest::bg, GNUNET_BLOCK_group_create(), GNUNET_BLOCK_group_destroy(), GNUNET_BLOCK_group_set_seen(), GNUNET_BLOCK_TYPE_FS_UBLOCK, GNUNET_break, GNUNET_OK, GSF_block_ctx, GSF_PendingRequest::replies_seen, GSF_PendingRequest::replies_seen_count, and type.

Referenced by GSF_pending_request_create_(), and GSF_pending_request_update_().

Here is the call graph for this function:

Here is the caller graph for this function:

GSF_pending_request_create_()

struct GSF_PendingRequest * GSF_pending_request_create_	(	enum GSF_PendingRequestOptions	options,
		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	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
	)

Create a new pending request.

Parameters

options	request options
type	type of the block that is being requested
query	key for the lookup
target	preferred target for the request, NULL for none
bf_data	raw data for bloom filter for known replies, can be NULL
bf_size	number of bytes in bf_data
anonymity_level	desired anonymity level
priority	maximum outgoing cumulative request priority to use
ttl	current time-to-live for the request
sender_pid	peer ID to use for the sender when forwarding, 0 for none; reference counter is taken over by this function
origin_pid	peer ID of origin of query (do not loop back)
replies_seen	hash codes of known local replies
replies_seen_count	size of the 'replies_seen' array
rh	handle to call when we get a reply
rh_cls	closure for rh

Returns

handle for the new pending request

Definition at line 278 of file gnunet-service-fs_pr.c.

{
  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);
#endif
  extra = 0;
  if (NULL != target)
    extra += sizeof(struct GNUNET_PeerIdentity);
  pr = GNUNET_malloc (sizeof(struct GSF_PendingRequest) + extra);
  pr->public_data.query = *query;
  eptr = (struct GNUNET_HashCode *) &pr[1];
  if (NULL != target)
  {
    pr->public_data.target = (struct GNUNET_PeerIdentity *) eptr;
    GNUNET_memcpy (eptr, target, sizeof(struct GNUNET_PeerIdentity));
  }
  pr->public_data.anonymity_level = anonymity_level;
  pr->public_data.priority = priority;
  pr->public_data.original_priority = priority;
  pr->public_data.options = options;
  pr->public_data.type = type;
  pr->public_data.start_time = GNUNET_TIME_absolute_get ();
  pr->sender_pid = sender_pid;
  pr->origin_pid = origin_pid;
  pr->rh = rh;
  pr->rh_cls = rh_cls;
  GNUNET_assert ((sender_pid != 0) || (0 == (options & GSF_PRO_FORWARD_ONLY)));
  if (ttl >= 0)
    pr->public_data.ttl = GNUNET_TIME_relative_to_absolute (
      GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, (uint32_t) ttl));
  else
    pr->public_data.ttl = GNUNET_TIME_absolute_subtract (
      pr->public_data.start_time,
      GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS,
                                     (uint32_t) (-ttl)));
  if (replies_seen_count > 0)
  {
    pr->replies_seen_size = replies_seen_count;
    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) &&
      (GNUNET_BLOCK_TYPE_FS_UBLOCK == pr->public_data.type))
  {
    pr->bg = GNUNET_BLOCK_group_create (GSF_block_ctx,
                                        pr->public_data.type,
                                        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->public_data.type, pr);
  }
  GNUNET_CONTAINER_multihashmap_put (pr_map,
                                     &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,
                                              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 (NULL != dpr);
      if (pr == dpr)
        break;     /* let the request live briefly... */
      if (NULL != dpr->rh)
        dpr->rh (dpr->rh_cls,
                 GNUNET_BLOCK_REPLY_TYPE_NOT_SUPPORTED,
                 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,
                            gettext_noop ("# Pending requests active"),
                            1,
                            GNUNET_NO);
  return pr;
}

References GNUNET_TIME_Absolute::abs_value_us, anonymity_level, GSF_PendingRequestData::anonymity_level, GSF_PendingRequest::bg, gettext_noop, GNUNET_assert, GNUNET_BLOCK_group_create(), GNUNET_BLOCK_REPLY_TYPE_NOT_SUPPORTED, GNUNET_BLOCK_TYPE_ANY, GNUNET_BLOCK_TYPE_FS_UBLOCK, GNUNET_CONTAINER_heap_get_size(), GNUNET_CONTAINER_heap_insert(), GNUNET_CONTAINER_heap_peek(), GNUNET_CONTAINER_multihashmap_put(), GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE, GNUNET_ERROR_TYPE_DEBUG, GNUNET_h2s(), GNUNET_log, GNUNET_malloc, GNUNET_memcpy, GNUNET_new_array, GNUNET_NO, GNUNET_STATISTICS_update(), GNUNET_TIME_absolute_get(), GNUNET_TIME_absolute_subtract(), GNUNET_TIME_relative_multiply(), GNUNET_TIME_relative_to_absolute(), GNUNET_TIME_UNIT_FOREVER_ABS, GNUNET_TIME_UNIT_SECONDS, GNUNET_YES, GSF_block_ctx, GSF_pending_request_cancel_(), GSF_PRO_BLOOMFILTER_FULL_REFRESH, GSF_PRO_FORWARD_ONLY, GSF_PRO_REQUEST_NEVER_EXPIRES, GSF_stats, GSF_PendingRequest::hnode, max_pending_requests, options, GSF_PendingRequestData::options, GSF_PendingRequest::origin_pid, GSF_PendingRequestData::original_priority, pr_map, GSF_PendingRequestData::priority, GSF_PendingRequest::public_data, GSF_PendingRequestData::query, refresh_bloomfilter(), GSF_PendingRequest::replies_seen, GSF_PendingRequest::replies_seen_count, GSF_PendingRequest::replies_seen_size, requests_by_expiration_heap, GSF_PendingRequest::rh, GSF_PendingRequest::rh_cls, GSF_PendingRequest::sender_pid, GSF_PendingRequestData::start_time, GSF_PendingRequestData::target, ttl, GSF_PendingRequestData::ttl, type, and GSF_PendingRequestData::type.

Referenced by handle_client_start_search(), and handle_p2p_get().

Here is the call graph for this function:

Here is the caller graph for this function:

GSF_pending_request_get_data_()

struct GSF_PendingRequestData * GSF_pending_request_get_data_

(

struct GSF_PendingRequest *

pr

)

Obtain the public data associated with a pending request.

Parameters

pr	pending request

Returns

associated public data

Definition at line 410 of file gnunet-service-fs_pr.c.

{
  return &pr->public_data;
}

References GSF_PendingRequest::public_data.

Referenced by client_response_handler(), free_pending_request(), get_latest(), get_rp_key(), GSF_plan_add_(), GSF_plan_notify_peer_disconnect_(), GSF_plan_notify_request_done_(), handle_client_start_search(), handle_p2p_get(), handle_p2p_reply(), merge_pr(), peer_request_destroy(), plan(), start_p2p_processing(), and test_exist_cb().

Here is the caller graph for this function:

GSF_pending_request_is_compatible_()

int GSF_pending_request_is_compatible_	(	struct GSF_PendingRequest *	pra,
		struct GSF_PendingRequest *	prb
	)

Test if two pending requests are compatible (would generate the same query modulo filters and should thus be processed jointly).

Parameters

pra	a pending request
prb	another pending request

Returns

GNUNET_OK if the requests are compatible

Definition at line 426 of file gnunet-service-fs_pr.c.

{
  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;
}

References GNUNET_NO, GNUNET_OK, GSF_PendingRequest::public_data, GSF_PendingRequestData::query, and GSF_PendingRequestData::type.

Referenced by merge_pr().

Here is the caller graph for this function:

GSF_pending_request_update_()

void GSF_pending_request_update_	(	struct GSF_PendingRequest *	pr,
		const struct GNUNET_HashCode *	replies_seen,
		unsigned int	replies_seen_count
	)

Update a given pending request with additional replies that have been seen.

Parameters

pr	request to update
replies_seen	hash codes of replies that we've seen
replies_seen_count	size of the replies_seen array

Definition at line 439 of file gnunet-service-fs_pr.c.

{
  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,
                         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);
    pr->replies_seen_count += replies_seen_count;
    refresh_bloomfilter (pr->public_data.type, pr);
  }
  else
  {
    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 */
      refresh_bloomfilter (pr->public_data.type, pr);
    }
    else
    {
      GNUNET_break (GNUNET_OK ==
                    GNUNET_BLOCK_group_set_seen (pr->bg,
                                                 replies_seen,
                                                 pr->replies_seen_count));
    }
  }
  if (NULL != pr->gh)
    GNUNET_DHT_get_filter_known_results (pr->gh,
                                         replies_seen_count,
                                         replies_seen);
}

References GSF_PendingRequest::bg, GSF_PendingRequest::gh, GNUNET_array_grow, GNUNET_BLOCK_group_set_seen(), GNUNET_break, GNUNET_DHT_get_filter_known_results(), GNUNET_memcpy, GNUNET_OK, GSF_PRO_BLOOMFILTER_FULL_REFRESH, GSF_PendingRequestData::options, GSF_PendingRequest::public_data, refresh_bloomfilter(), GSF_PendingRequest::replies_seen, GSF_PendingRequest::replies_seen_count, GSF_PendingRequest::replies_seen_size, and GSF_PendingRequestData::type.

Referenced by handle_client_start_search(), and process_reply().

Here is the call graph for this function:

Here is the caller graph for this function:

GSF_pending_request_get_message_()

struct GNUNET_MQ_Envelope * GSF_pending_request_get_message_

(

struct GSF_PendingRequest *

pr

)

Generate the message corresponding to the given pending request for transmission to other peers.

Parameters

pr	request to generate the message for

Returns

envelope with the request message

Definition at line 489 of file gnunet-service-fs_pr.c.

{
  struct GNUNET_MQ_Envelope *env;
  struct GetMessage *gm;
  struct GNUNET_PeerIdentity *ext;
  unsigned int k;
  uint32_t bm;
  uint32_t prio;
  size_t bf_size;
  struct GNUNET_TIME_Absolute now;
  int64_t ttl;
  int do_route;
  void *bf_data;
 
  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));
  if ((! do_route) && (pr->sender_pid == 0))
  {
    GNUNET_break (0);
    do_route = GNUNET_YES;
  }
  if (! do_route)
  {
    bm |= GET_MESSAGE_BIT_RETURN_TO;
    k++;
  }
  if (NULL != pr->public_data.target)
  {
    bm |= GET_MESSAGE_BIT_TRANSMIT_TO;
    k++;
  }
  if (GNUNET_OK !=
      GNUNET_BLOCK_group_serialize (pr->bg,
                                    &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 = GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
                                     pr->public_data.priority + 1);
  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_us - now.abs_value_us);
  gm->ttl = htonl (ttl / 1000LL / 1000LL);
  gm->reserved = htonl (0);
  gm->hash_bitmap = htonl (bm);
  gm->query = pr->public_data.query;
  ext = (struct GNUNET_PeerIdentity *) &gm[1];
  k = 0;
  if (! do_route)
    GNUNET_PEER_resolve (pr->sender_pid, &ext[k++]);
  if (NULL != pr->public_data.target)
    ext[k++] = *pr->public_data.target;
  GNUNET_memcpy (&ext[k], bf_data, bf_size);
  GNUNET_free (bf_data);
  return env;
}

References GNUNET_TIME_Absolute::abs_value_us, GSF_PendingRequest::bg, env, GET_MESSAGE_BIT_RETURN_TO, GET_MESSAGE_BIT_TRANSMIT_TO, GNUNET_BLOCK_group_serialize(), GNUNET_break, GNUNET_CRYPTO_QUALITY_WEAK, GNUNET_CRYPTO_random_u32(), GNUNET_ERROR_TYPE_DEBUG, GNUNET_free, GNUNET_h2s(), GNUNET_log, GNUNET_memcpy, GNUNET_MESSAGE_TYPE_FS_GET, GNUNET_MQ_msg_extra, GNUNET_OK, GNUNET_PEER_resolve(), GNUNET_TIME_absolute_get(), GNUNET_YES, GSF_PRO_FORWARD_ONLY, GetMessage::hash_bitmap, GSF_PendingRequestData::num_transmissions, GSF_PendingRequestData::options, GetMessage::priority, GSF_PendingRequestData::priority, GSF_PendingRequest::public_data, GetMessage::query, GSF_PendingRequestData::query, GetMessage::reserved, GSF_PendingRequestData::respect_offered, GSF_PendingRequest::sender_pid, GSF_PendingRequestData::target, ttl, GetMessage::ttl, GSF_PendingRequestData::ttl, GetMessage::type, and GSF_PendingRequestData::type.

Referenced by schedule_peer_transmission().

Here is the call graph for this function:

Here is the caller graph for this function:

clean_request()

static int clean_request ( void *  cls, const struct GNUNET_HashCode *  key, void *  value  )

static

Iterator to free pending requests.

Parameters

cls	closure, unused
key	current key code
value	value in the hash map (pending request)

Returns

GNUNET_YES (we should continue to iterate)

Definition at line 573 of file gnunet-service-fs_pr.c.

{
  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->cadet_request)
  {
    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);
  }
  GSF_plan_notify_request_done_ (pr);
  GNUNET_free (pr->replies_seen);
  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);
  pr->origin_pid = 0;
  if (NULL != pr->hnode)
  {
    GNUNET_CONTAINER_heap_remove_node (pr->hnode);
    pr->hnode = NULL;
  }
  if (NULL != pr->qe)
  {
    GNUNET_DATASTORE_cancel (pr->qe);
    pr->qe = NULL;
  }
  if (NULL != pr->gh)
  {
    GNUNET_DHT_get_stop (pr->gh);
    pr->gh = NULL;
  }
  if (NULL != pr->warn_task)
  {
    GNUNET_SCHEDULER_cancel (pr->warn_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,
                            GNUNET_NO);
  GNUNET_free (pr);
  return GNUNET_YES;
}

References GSF_PendingRequest::bg, GSF_PendingRequest::cadet_request, GSF_PendingRequest::cadet_retry_count, CADET_RETRY_MAX, gettext_noop, GSF_PendingRequest::gh, GNUNET_assert, GNUNET_BLOCK_group_destroy(), GNUNET_CONTAINER_heap_remove_node(), GNUNET_CONTAINER_multihashmap_remove(), GNUNET_DATASTORE_cancel(), GNUNET_DHT_get_stop(), GNUNET_ERROR_TYPE_DEBUG, GNUNET_free, GNUNET_h2s(), GNUNET_log, GNUNET_NO, GNUNET_OK, GNUNET_PEER_change_rc(), GNUNET_SCHEDULER_cancel(), GNUNET_STATISTICS_update(), GNUNET_YES, GSF_cadet_query_cancel(), GSF_plan_notify_request_done_(), GSF_stats, GSF_PendingRequest::hnode, key, GSF_PendingRequest::llc_cont, GSF_PendingRequest::llc_cont_cls, GSF_PendingRequest::local_result, GSF_PendingRequest::origin_pid, pr_map, GSF_PendingRequest::public_data, GSF_PendingRequest::qe, GSF_PendingRequestData::query, GSF_PendingRequest::replies_seen, GSF_PendingRequest::sender_pid, value, and GSF_PendingRequest::warn_task.

Referenced by GSF_pending_request_cancel_(), and GSF_pending_request_done_().

Here is the call graph for this function:

Here is the caller graph for this function:

GSF_pending_request_cancel_()

void GSF_pending_request_cancel_	(	struct GSF_PendingRequest *	pr,
		int	full_cleanup
	)

Explicitly cancel a pending request.

Parameters

pr	request to cancel
full_cleanup	fully purge the request

Definition at line 641 of file gnunet-service-fs_pr.c.

{
  GSF_LocalLookupContinuation cont;
 
  if (NULL == pr_map)
    return; /* already cleaned up! */
  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->cadet_request)
    {
      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);
    }
    GSF_plan_notify_request_done_ (pr);
    if (NULL != pr->qe)
    {
      GNUNET_DATASTORE_cancel (pr->qe);
      pr->qe = NULL;
    }
    if (NULL != pr->gh)
    {
      GNUNET_DHT_get_stop (pr->gh);
      pr->gh = NULL;
    }
    if (NULL != pr->warn_task)
    {
      GNUNET_SCHEDULER_cancel (pr->warn_task);
      pr->warn_task = NULL;
    }
    return;
  }
  GNUNET_assert (GNUNET_YES ==
                 clean_request (NULL, &pr->public_data.query, pr));
}

References GSF_PendingRequest::cadet_request, GSF_PendingRequest::cadet_retry_count, CADET_RETRY_MAX, clean_request(), GSF_PendingRequest::gh, GNUNET_assert, GNUNET_DATASTORE_cancel(), GNUNET_DHT_get_stop(), GNUNET_NO, GNUNET_SCHEDULER_cancel(), GNUNET_YES, GSF_cadet_query_cancel(), GSF_plan_notify_request_done_(), GSF_PendingRequest::llc_cont, GSF_PendingRequest::llc_cont_cls, GSF_PendingRequest::local_result, pr_map, GSF_PendingRequest::public_data, GSF_PendingRequest::qe, GSF_PendingRequestData::query, GSF_PendingRequest::rh, and GSF_PendingRequest::warn_task.

Referenced by cancel_pending_request(), client_request_destroy(), GSF_pending_request_create_(), and test_exist_cb().

Here is the call graph for this function:

Here is the caller graph for this function:

GSF_iterate_pending_requests_()

void GSF_iterate_pending_requests_	(	GSF_PendingRequestIterator	it,
		void *	cls
	)

Iterate over all pending requests.

Parameters

it	function to call for each request
cls	closure for it

Definition at line 690 of file gnunet-service-fs_pr.c.

{
  GNUNET_CONTAINER_multihashmap_iterate (
    pr_map,
    (GNUNET_CONTAINER_MultiHashMapIteratorCallback) it,
    cls);
}

References GNUNET_CONTAINER_multihashmap_iterate(), and pr_map.

Referenced by GSF_peer_connect_handler().

Here is the call graph for this function:

Here is the caller graph for this function:

update_request_performance_data()

static void update_request_performance_data ( struct ProcessReplyClosure *  prq, struct GSF_PendingRequest *  pr  )

static

Update the performance data for the sender (if any) since the sender successfully answered one of our queries.

Parameters

prq	information about the sender
pr	request that was satisfied

Definition at line 759 of file gnunet-service-fs_pr.c.

{
  if (prq->sender == NULL)
    return;
  GSF_peer_update_performance_ (prq->sender,
                                pr->public_data.start_time,
                                prq->priority);
}

References GSF_peer_update_performance_(), ProcessReplyClosure::priority, GSF_PendingRequest::public_data, ProcessReplyClosure::sender, and GSF_PendingRequestData::start_time.

Referenced by process_reply().

Here is the call graph for this function:

Here is the caller graph for this function:

process_reply()

static enum GNUNET_GenericReturnValue process_reply ( void *  cls, const struct GNUNET_HashCode *  key, void *  value  )

static

We have received a reply; handle it!

Parameters

cls	response (a struct ProcessReplyClosure)
key	our query
value	value in the hash map (info about the query)

Returns

GNUNET_YES (we should continue to iterate)

Definition at line 779 of file gnunet-service-fs_pr.c.

{
  struct ProcessReplyClosure *prq = cls;
  struct GSF_PendingRequest *pr = value;
  struct GNUNET_HashCode chash;
  struct GNUNET_TIME_Absolute last_transmission;
 
  if (NULL == pr->rh)
    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));
  GNUNET_STATISTICS_update (GSF_stats,
                            gettext_noop ("# replies received and matched"),
                            1,
                            GNUNET_NO);
  prq->eval = GNUNET_BLOCK_check_reply (GSF_block_ctx,
                                        prq->type,
                                        pr->bg,
                                        key,
                                        NULL, 0,
                                        prq->data,
                                        prq->size);
  switch (prq->eval)
  {
  case GNUNET_BLOCK_REPLY_OK_MORE:
    update_request_performance_data (prq, pr);
    break;
  case GNUNET_BLOCK_REPLY_OK_LAST:
    /* short cut: stop processing early, no BF-update, etc. */
    update_request_performance_data (prq, pr);
    GNUNET_LOAD_update (GSF_rt_entry_lifetime,
                        GNUNET_TIME_absolute_get_duration (
                          pr->public_data.start_time)
                        .rel_value_us);
    if (GNUNET_YES !=
        GSF_request_plan_reference_get_last_transmission_ (pr->public_data
                                                           .pr_head,
                                                           prq->sender,
                                                           &last_transmission))
      last_transmission = GNUNET_TIME_UNIT_FOREVER_ABS;
    /* pass on to other peers / local clients */
    pr->rh (pr->rh_cls,
            prq->eval,
            pr,
            prq->anonymity_level,
            prq->expiration,
            last_transmission,
            prq->type,
            prq->data,
            prq->size);
    return GNUNET_YES;
  case GNUNET_BLOCK_REPLY_OK_DUPLICATE:
#if INSANE_STATISTICS
    GNUNET_STATISTICS_update (GSF_stats,
                              "# duplicate replies discarded (bloomfilter)",
                              1,
                              GNUNET_NO);
#endif
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Duplicate response, discarding.\n");
    return GNUNET_YES;   /* duplicate */
  case GNUNET_BLOCK_REPLY_IRRELEVANT:
    GNUNET_STATISTICS_update (GSF_stats,
                              "# irrelevant replies discarded",
                              1,
                              GNUNET_NO);
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Irrelevant response, ignoring.\n");
    return GNUNET_YES;
  case GNUNET_BLOCK_REPLY_TYPE_NOT_SUPPORTED:
    GNUNET_break (0); /* bad installation? */
    return GNUNET_NO;
  }
  /* update bloomfilter */
  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,
                              GNUNET_NO);
  }
  else
  {
    GSF_dht_lookup_ (pr);
  }
  prq->priority += pr->public_data.original_priority;
  pr->public_data.priority = 0;
  pr->public_data.original_priority = 0;
  pr->public_data.results_found++;
  prq->request_found = GNUNET_YES;
  /* finally, pass on to other peer / local client */
  if (! GSF_request_plan_reference_get_last_transmission_ (pr->public_data
                                                           .pr_head,
                                                           prq->sender,
                                                           &last_transmission))
    last_transmission = GNUNET_TIME_UNIT_FOREVER_ABS;
  pr->rh (pr->rh_cls,
          prq->eval,
          pr,
          prq->anonymity_level,
          prq->expiration,
          last_transmission,
          prq->type,
          prq->data,
          prq->size);
  return GNUNET_YES;
}

References ProcessReplyClosure::anonymity_level, GSF_PendingRequest::bg, ProcessReplyClosure::data, ProcessReplyClosure::eval, ProcessReplyClosure::expiration, gettext_noop, GNUNET_BLOCK_check_reply(), GNUNET_BLOCK_REPLY_IRRELEVANT, GNUNET_BLOCK_REPLY_OK_DUPLICATE, GNUNET_BLOCK_REPLY_OK_LAST, GNUNET_BLOCK_REPLY_OK_MORE, GNUNET_BLOCK_REPLY_TYPE_NOT_SUPPORTED, GNUNET_break, GNUNET_CRYPTO_hash(), GNUNET_ERROR_TYPE_DEBUG, GNUNET_h2s(), GNUNET_LOAD_update(), GNUNET_log, GNUNET_NO, GNUNET_STATISTICS_update(), GNUNET_TIME_absolute_get_duration(), GNUNET_TIME_UNIT_FOREVER_ABS, GNUNET_YES, GSF_block_ctx, GSF_dht_lookup_(), GSF_pending_request_update_(), GSF_request_plan_reference_get_last_transmission_(), GSF_rt_entry_lifetime, GSF_stats, key, GSF_PendingRequestData::original_priority, GSF_PendingRequestData::pr_head, ProcessReplyClosure::priority, GSF_PendingRequestData::priority, GSF_PendingRequest::public_data, GNUNET_TIME_Relative::rel_value_us, ProcessReplyClosure::request_found, GSF_PendingRequestData::results_found, GSF_PendingRequest::rh, GSF_PendingRequest::rh_cls, ProcessReplyClosure::sender, ProcessReplyClosure::size, GSF_PendingRequestData::start_time, ProcessReplyClosure::type, update_request_performance_data(), and value.

Referenced by cadet_reply_proc(), handle_dht_reply(), handle_p2p_put(), and process_local_reply().

Here is the call graph for this function:

Here is the caller graph for this function:

put_migration_continuation()

static void put_migration_continuation ( void *  cls, int  success, struct GNUNET_TIME_Absolute  min_expiration, const char *  msg  )

static

Continuation called to notify client about result of the operation.

Parameters

cls	closure
success	GNUNET_SYSERR on failure
min_expiration	minimum expiration time required for content to be stored
msg	NULL on success, otherwise an error message

Definition at line 934 of file gnunet-service-fs_pr.c.

{
  struct PutMigrationContext *pmc = cls;
  struct GSF_ConnectedPeer *cp;
  struct GNUNET_TIME_Relative mig_pause;
  struct GSF_PeerPerformanceData *ppd;
 
  if (NULL != datastore_put_load)
  {
    if (GNUNET_SYSERR != success)
    {
      GNUNET_LOAD_update (datastore_put_load,
                          GNUNET_TIME_absolute_get_duration (pmc->start)
                          .rel_value_us);
    }
    else
    {
      /* on queue failure / timeout, increase the put load dramatically */
      GNUNET_LOAD_update (datastore_put_load,
                          GNUNET_TIME_UNIT_MINUTES.rel_value_us);
    }
  }
  cp = GSF_peer_get_ (&pmc->origin);
  if (GNUNET_OK == success)
  {
    if (NULL != cp)
    {
      ppd = GSF_get_peer_performance_data_ (cp);
      ppd->migration_delay.rel_value_us /= 2;
    }
    GNUNET_free (pmc);
    return;
  }
  if ((GNUNET_NO == success) && (GNUNET_NO == pmc->requested) && (NULL != cp))
  {
    ppd = GSF_get_peer_performance_data_ (cp);
    if (min_expiration.abs_value_us > 0)
    {
      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                  "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 = GNUNET_TIME_relative_max (GNUNET_TIME_UNIT_SECONDS,
                                                       ppd->migration_delay);
      ppd->migration_delay =
        GNUNET_TIME_relative_min (GNUNET_TIME_UNIT_HOURS, 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_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_free (pmc);
  GNUNET_STATISTICS_update (GSF_stats,
                            gettext_noop ("# Datastore `PUT' failures"),
                            1,
                            GNUNET_NO);
}

References GNUNET_TIME_Absolute::abs_value_us, datastore_put_load, gettext_noop, GNUNET_CRYPTO_QUALITY_WEAK, GNUNET_CRYPTO_random_u64(), GNUNET_ERROR_TYPE_DEBUG, GNUNET_free, GNUNET_LOAD_update(), GNUNET_log, GNUNET_NO, GNUNET_OK, GNUNET_STATISTICS_update(), GNUNET_STRINGS_relative_time_to_string(), GNUNET_SYSERR, GNUNET_TIME_absolute_get_duration(), GNUNET_TIME_absolute_get_remaining(), GNUNET_TIME_relative_max(), GNUNET_TIME_relative_min(), GNUNET_TIME_relative_saturating_multiply(), GNUNET_TIME_relative_to_absolute(), GNUNET_TIME_UNIT_HOURS, GNUNET_TIME_UNIT_MINUTES, GNUNET_TIME_UNIT_SECONDS, GNUNET_YES, GSF_block_peer_migration_(), GSF_get_peer_performance_data_(), GSF_peer_get_(), GSF_stats, GSF_PeerPerformanceData::migration_delay, min_expiration, PutMigrationContext::origin, GNUNET_TIME_Relative::rel_value_us, PutMigrationContext::requested, and PutMigrationContext::start.

Referenced by handle_dht_reply(), and handle_p2p_put().

Here is the call graph for this function:

Here is the caller graph for this function:

test_put_load_too_high()

static int test_put_load_too_high ( uint32_t  priority )

static

Test if the DATABASE (PUT) load on this peer is too high to even consider processing the query at all.

Parameters

priority

the priority of the item

Returns

GNUNET_YES if the load is too high to do anything (load high) GNUNET_NO to process normally (load normal or low)

Definition at line 1019 of file gnunet-service-fs_pr.c.

{
  double ld;
 
  if (NULL == datastore_put_load)
    return GNUNET_NO;
  if (GNUNET_LOAD_get_average (datastore_put_load) < 50)
    return GNUNET_NO; /* very fast */
  ld = GNUNET_LOAD_get_load (datastore_put_load);
  if (ld < 2.0 * (1 + priority))
    return GNUNET_NO;
  GNUNET_STATISTICS_update (GSF_stats,
                            gettext_noop (
                              "# storage requests dropped due to high load"),
                            1,
                            GNUNET_NO);
  return GNUNET_YES;
}

References datastore_put_load, gettext_noop, GNUNET_LOAD_get_average(), GNUNET_LOAD_get_load(), GNUNET_NO, GNUNET_STATISTICS_update(), GNUNET_YES, and GSF_stats.

Referenced by handle_dht_reply(), and handle_p2p_put().

Here is the call graph for this function:

Here is the caller graph for this function:

handle_dht_reply()

static void handle_dht_reply ( void *  cls, struct GNUNET_TIME_Absolute  exp, const struct GNUNET_HashCode *  key, const struct GNUNET_PeerIdentity *  trunc_peer, const struct GNUNET_DHT_PathElement *  get_path, unsigned int  get_path_length, const struct GNUNET_DHT_PathElement *  put_path, unsigned int  put_path_length, enum GNUNET_BLOCK_Type  type, size_t  size, const void *  data  )

static

Iterator called on each result obtained for a DHT operation that expects a reply.

Parameters

cls	closure
exp	when will this value expire
key	key of the result
trunc_peer	truncated peer, NULL for none
get_path	peers on reply path (or NULL if not recorded)
get_path_length	number of entries in get_path
put_path	peers on the PUT path (or NULL if not recorded)
put_path_length	number of entries in get_path
type	type of the result
size	number of bytes in data
data	pointer to the result data

Definition at line 1056 of file gnunet-service-fs_pr.c.

{
  struct GSF_PendingRequest *pr = cls;
  struct ProcessReplyClosure prq;
  struct PutMigrationContext *pmc;
 
  GNUNET_STATISTICS_update (GSF_stats,
                            gettext_noop ("# Replies received from DHT"),
                            1,
                            GNUNET_NO);
  memset (&prq, 0, sizeof(prq));
  prq.data = data;
  prq.expiration = exp;
  /* do not allow migrated content to live longer than 1 year */
  prq.expiration = GNUNET_TIME_absolute_min (GNUNET_TIME_relative_to_absolute (
                                               GNUNET_TIME_UNIT_YEARS),
                                             prq.expiration);
  prq.size = size;
  prq.type = type;
  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_new (struct PutMigrationContext);
    pmc->start = GNUNET_TIME_absolute_get ();
    pmc->requested = GNUNET_YES;
    if (NULL == GNUNET_DATASTORE_put (GSF_dsh,
                                      0,
                                      key,
                                      size,
                                      data,
                                      type,
                                      prq.priority,
                                      1 /* anonymity */,
                                      0 /* replication */,
                                      exp,
                                      1 + prq.priority,
                                      MAX_DATASTORE_QUEUE,
                                      &put_migration_continuation,
                                      pmc))
    {
      put_migration_continuation (pmc,
                                  GNUNET_SYSERR,
                                  GNUNET_TIME_UNIT_ZERO_ABS,
                                  NULL);
    }
  }
}

References active_to_migration, data, ProcessReplyClosure::data, ProcessReplyClosure::expiration, gettext_noop, GNUNET_DATASTORE_put(), GNUNET_ERROR_TYPE_DEBUG, GNUNET_h2s(), GNUNET_log, GNUNET_new, GNUNET_NO, GNUNET_STATISTICS_update(), GNUNET_SYSERR, GNUNET_TIME_absolute_get(), GNUNET_TIME_absolute_min(), GNUNET_TIME_relative_to_absolute(), GNUNET_TIME_UNIT_YEARS, GNUNET_TIME_UNIT_ZERO_ABS, GNUNET_YES, GSF_dsh, GSF_stats, key, MAX_DATASTORE_QUEUE, ProcessReplyClosure::priority, process_reply(), put_migration_continuation(), PutMigrationContext::requested, size, ProcessReplyClosure::size, PutMigrationContext::start, test_put_load_too_high(), type, and ProcessReplyClosure::type.

Referenced by GSF_dht_lookup_().

Here is the call graph for this function:

Here is the caller graph for this function:

GSF_dht_lookup_()

void GSF_dht_lookup_

(

struct GSF_PendingRequest *

pr

)

Consider looking up the data in the DHT (anonymity-level permitting).

Parameters

pr	the pending request to process

Definition at line 1128 of file gnunet-service-fs_pr.c.

{
  const void *xquery;
  size_t xquery_size;
  struct GNUNET_PeerIdentity pi;
  char buf[sizeof(struct GNUNET_HashCode) * 2] GNUNET_ALIGN;
 
  if (0 != pr->public_data.anonymity_level)
    return;
  if (NULL != pr->gh)
  {
    GNUNET_DHT_get_stop (pr->gh);
    pr->gh = NULL;
  }
  xquery = NULL;
  xquery_size = 0;
  if (0 != (pr->public_data.options & GSF_PRO_FORWARD_ONLY))
  {
    GNUNET_assert (0 != pr->sender_pid);
    GNUNET_PEER_resolve (pr->sender_pid, &pi);
    GNUNET_memcpy (&buf[xquery_size], &pi, sizeof(struct GNUNET_PeerIdentity));
    xquery_size += sizeof(struct GNUNET_PeerIdentity);
  }
  pr->gh = GNUNET_DHT_get_start (GSF_dht,
                                 pr->public_data.type,
                                 &pr->public_data.query,
                                 DHT_GET_REPLICATION,
                                 GNUNET_DHT_RO_DEMULTIPLEX_EVERYWHERE,
                                 xquery,
                                 xquery_size,
                                 &handle_dht_reply,
                                 pr);
  if ((NULL != pr->gh) && (0 != pr->replies_seen_count))
    GNUNET_DHT_get_filter_known_results (pr->gh,
                                         pr->replies_seen_count,
                                         pr->replies_seen);
}

References GSF_PendingRequestData::anonymity_level, DHT_GET_REPLICATION, GSF_PendingRequest::gh, GNUNET_ALIGN, GNUNET_assert, GNUNET_DHT_get_filter_known_results(), GNUNET_DHT_get_start(), GNUNET_DHT_get_stop(), GNUNET_DHT_RO_DEMULTIPLEX_EVERYWHERE, GNUNET_memcpy, GNUNET_PEER_resolve(), GSF_dht, GSF_PRO_FORWARD_ONLY, handle_dht_reply(), GSF_PendingRequestData::options, GSF_PendingRequest::public_data, GSF_PendingRequestData::query, GSF_PendingRequest::replies_seen, GSF_PendingRequest::replies_seen_count, GSF_PendingRequest::sender_pid, and GSF_PendingRequestData::type.

Referenced by process_reply(), and start_p2p_processing().

Here is the call graph for this function:

Here is the caller graph for this function:

cadet_reply_proc()

static void cadet_reply_proc ( void *  cls, enum GNUNET_BLOCK_Type  type, struct GNUNET_TIME_Absolute  expiration, size_t  data_size, const void *  data  )

static

Function called with a reply from the cadet.

Parameters

cls	the pending request struct
type	type of the block, ANY on error
expiration	expiration time for the block
data_size	number of bytes in data, 0 on error
data	reply block data, NULL on error

Definition at line 1177 of file gnunet-service-fs_pr.c.

{
  struct GSF_PendingRequest *pr = cls;
  struct ProcessReplyClosure prq;
  struct GNUNET_HashCode query;
 
  pr->cadet_request = NULL;
  if (GNUNET_OK !=
      GNUNET_BLOCK_check_block (GSF_block_ctx,
                                type,
                                data,
                                data_size))
  {
    GNUNET_break_op (0);
    return;
  }
  if (GNUNET_BLOCK_TYPE_ANY == type)
  {
    GNUNET_break (NULL == data);
    GNUNET_break (0 == data_size);
    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 cadet\n");
    /* retry -- without delay, as this is non-anonymous
       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 !=
      GNUNET_BLOCK_get_key (GSF_block_ctx,
                            type,
                            data,
                            data_size,
                            &query))
  {
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Failed to derive key for block of type %d\n",
                (int) type);
    GNUNET_break_op (0);
    return;
  }
  GNUNET_STATISTICS_update (GSF_stats,
                            gettext_noop ("# Replies received from CADET"),
                            1,
                            GNUNET_NO);
  memset (&prq, 0, sizeof(prq));
  prq.data = data;
  prq.expiration = expiration;
  /* do not allow migrated content to live longer than 1 year */
  prq.expiration = GNUNET_TIME_absolute_min (GNUNET_TIME_relative_to_absolute (
                                               GNUNET_TIME_UNIT_YEARS),
                                             prq.expiration);
  prq.size = data_size;
  prq.type = type;
  process_reply (&prq,
                 &query,
                 pr);
}

References cadet_reply_proc(), GSF_PendingRequest::cadet_request, GSF_PendingRequest::cadet_retry_count, CADET_RETRY_MAX, data, ProcessReplyClosure::data, data_size, expiration, ProcessReplyClosure::expiration, gettext_noop, GNUNET_BLOCK_check_block(), GNUNET_BLOCK_get_key(), GNUNET_BLOCK_TYPE_ANY, GNUNET_break, GNUNET_break_op, GNUNET_ERROR_TYPE_DEBUG, GNUNET_log, GNUNET_NO, GNUNET_OK, GNUNET_STATISTICS_update(), GNUNET_TIME_absolute_min(), GNUNET_TIME_relative_to_absolute(), GNUNET_TIME_UNIT_YEARS, GNUNET_YES, GSF_block_ctx, GSF_cadet_query(), GSF_stats, process_reply(), GSF_PendingRequest::public_data, GSF_PendingRequestData::query, ProcessReplyClosure::size, GSF_PendingRequestData::target, type, ProcessReplyClosure::type, and GSF_PendingRequestData::type.

Referenced by cadet_reply_proc(), and GSF_cadet_lookup_().

Here is the call graph for this function:

Here is the caller graph for this function:

GSF_cadet_lookup_()

void GSF_cadet_lookup_

(

struct GSF_PendingRequest *

pr

)

Consider downloading via cadet (if possible)

Parameters

pr	the pending request to process

Definition at line 1252 of file gnunet-service-fs_pr.c.

{
  if (0 != pr->public_data.anonymity_level)
    return;
  if (0 == pr->public_data.target)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Cannot do cadet-based download, target peer not known\n");
    return;
  }
  if (NULL != pr->cadet_request)
    return;
  pr->cadet_request = GSF_cadet_query (pr->public_data.target,
                                       &pr->public_data.query,
                                       pr->public_data.type,
                                       &cadet_reply_proc,
                                       pr);
}

References GSF_PendingRequestData::anonymity_level, cadet_reply_proc(), GSF_PendingRequest::cadet_request, GNUNET_ERROR_TYPE_DEBUG, GNUNET_log, GSF_cadet_query(), GSF_PendingRequest::public_data, GSF_PendingRequestData::query, GSF_PendingRequestData::target, and GSF_PendingRequestData::type.

Referenced by start_p2p_processing().

Here is the call graph for this function:

Here is the caller graph for this function:

warn_delay_task()

static void warn_delay_task ( void *  cls )

static

Task that issues a warning if the datastore lookup takes too long.

Parameters

cls	the struct GSF_PendingRequest

Definition at line 1278 of file gnunet-service-fs_pr.c.

{
  struct GSF_PendingRequest *pr = cls;
 
  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));
  pr->warn_task = GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_MINUTES,
                                                &warn_delay_task,
                                                pr);
}

References _, GNUNET_ERROR_TYPE_BULK, GNUNET_ERROR_TYPE_WARNING, GNUNET_log, GNUNET_SCHEDULER_add_delayed(), GNUNET_STRINGS_relative_time_to_string(), GNUNET_TIME_absolute_get_duration(), GNUNET_TIME_UNIT_MINUTES, GNUNET_YES, GSF_PendingRequest::qe_start, warn_delay_task(), and GSF_PendingRequest::warn_task.

Referenced by start_local_query(), and warn_delay_task().

Here is the call graph for this function:

Here is the caller graph for this function:

odc_warn_delay_task()

static void odc_warn_delay_task ( void *  cls )

static

Task that issues a warning if the datastore lookup takes too long.

Parameters

cls	the struct GSF_PendingRequest

Definition at line 1299 of file gnunet-service-fs_pr.c.

{
  struct GSF_PendingRequest *pr = cls;
 
  GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
              _ ("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);
}

References _, GNUNET_ERROR_TYPE_WARNING, GNUNET_log, GNUNET_SCHEDULER_add_delayed(), GNUNET_STRINGS_relative_time_to_string(), GNUNET_TIME_absolute_get_duration(), GNUNET_TIME_UNIT_MINUTES, GNUNET_YES, odc_warn_delay_task(), GSF_PendingRequest::qe_start, and GSF_PendingRequest::warn_task.

Referenced by odc_warn_delay_task(), and process_local_reply().

Here is the call graph for this function:

Here is the caller graph for this function:

call_continuation()

static void call_continuation ( struct GSF_PendingRequest *  pr )

static

Definition at line 1316 of file gnunet-service-fs_pr.c.

{
  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_REPLY_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_REPLY_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_REPLY_OK_LAST);
    return;
  }
 
  cont (pr->llc_cont_cls,
        pr,
        pr->local_result);
}

References GNUNET_assert, GNUNET_BLOCK_REPLY_OK_LAST, GNUNET_BLOCK_TYPE_ANY, GNUNET_SCHEDULER_cancel(), GNUNET_TIME_UNIT_FOREVER_ABS, GNUNET_TIME_UNIT_ZERO_ABS, GSF_PRO_LOCAL_ONLY, GSF_PendingRequest::llc_cont, GSF_PendingRequest::llc_cont_cls, GSF_PendingRequest::local_result, GSF_PendingRequestData::options, GSF_PendingRequest::public_data, GSF_PendingRequest::qe, GSF_PendingRequest::rh, GSF_PendingRequest::rh_cls, and GSF_PendingRequest::warn_task.

Referenced by no_more_local_results(), process_local_reply(), and start_local_query().

Here is the call graph for this function:

Here is the caller graph for this function:

no_more_local_results()

static void no_more_local_results ( struct GSF_PendingRequest *  pr )

static

Definition at line 1362 of file gnunet-service-fs_pr.c.

{
  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);
}

References call_continuation(), gettext_noop, GNUNET_BLOCK_TYPE_FS_DBLOCK, GNUNET_BLOCK_TYPE_FS_IBLOCK, GNUNET_ERROR_TYPE_BULK, GNUNET_ERROR_TYPE_DEBUG, GNUNET_log, GNUNET_NO, GNUNET_STATISTICS_update(), GSF_stats, GSF_PendingRequest::public_data, and GSF_PendingRequestData::type.

Referenced by process_local_reply().

Here is the call graph for this function:

Here is the caller graph for this function:

process_local_reply()

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  )

static

We're processing (local) results for a search request from another peer.

Pass applicable results to the peer and if we are done either clean up (operation complete) or forward to other peers (more results possible).

Parameters

cls	our closure (struct GSF_PendingRequest *)
key	key for the content
size	number of bytes in data
data	content stored
type	type of the content
priority	priority of the content
anonymity	anonymity-level for the content
replication	replication-level for the content
expiration	expiration time for the content
uid	unique identifier for the datum; maybe 0 if no unique identifier is available

Definition at line 1459 of file gnunet-service-fs_pr.c.

{
  struct GSF_PendingRequest *pr = cls;
  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)
    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 */
  {
    /* No results */
#if INSANE_STATISTICS
    GNUNET_STATISTICS_update (GSF_stats,
                              gettext_noop (
                                "# Datastore lookups concluded (no results)"),
                              1,
                              GNUNET_NO);
#endif
    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_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 (GNUNET_BLOCK_TYPE_FS_ONDEMAND == type)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Found ONDEMAND block, performing on-demand encoding\n");
    GNUNET_STATISTICS_update (GSF_stats,
                              gettext_noop (
                                "# on-demand blocks matched requests"),
                              1,
                              GNUNET_NO);
    pr->qe_start = GNUNET_TIME_absolute_get ();
    pr->warn_task = 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,
                                                       replication,
                                                       expiration,
                                                       uid,
                                                       &process_local_reply,
                                                       pr))
    {
      GNUNET_STATISTICS_update (GSF_stats,
                                gettext_noop (
                                  "# on-demand lookups performed successfully"),
                                1,
                                GNUNET_NO);
      return;     /* we're done */
    }
    GNUNET_STATISTICS_update (GSF_stats,
                              gettext_noop ("# on-demand lookups failed"),
                              1,
                              GNUNET_NO);
    GNUNET_SCHEDULER_cancel (pr->warn_task);
    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;
  prq.expiration = expiration;
  prq.size = size;
  if (GNUNET_OK !=
      GNUNET_BLOCK_get_key (GSF_block_ctx,
                            type,
                            data,
                            size,
                            &query))
  {
    GNUNET_break (0);
    GNUNET_DATASTORE_remove (GSF_dsh,
                             key,
                             size,
                             data,
                             UINT_MAX,
                             UINT_MAX,
                             NULL,
                             NULL);
    start_local_query (pr, uid + 1 /* next_uid */, false /* random */);
    return;
  }
  prq.type = type;
  prq.priority = priority;
  prq.request_found = GNUNET_NO;
  prq.anonymity_level = anonymity;
  if ((0 == old_rf) && (0 == pr->public_data.results_found))
    GSF_update_datastore_delay_ (pr->public_data.start_time);
  process_reply (&prq,
                 key,
                 pr);
  pr->local_result = prq.eval;
  if (GNUNET_BLOCK_REPLY_OK_LAST == prq.eval)
  {
    GNUNET_STATISTICS_update (
      GSF_stats,
      gettext_noop ("# Datastore lookups concluded (found last result)"),
      1,
      GNUNET_NO);
    call_continuation (pr);
    return;
  }
  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_STATISTICS_update (GSF_stats,
                              gettext_noop (
                                "# Datastore lookups concluded (load too high)"),
                              1,
                              GNUNET_NO);
    call_continuation (pr);
    return;
  }
  start_local_query (pr, uid + 1 /* next_uid */, false /* random */);
}

References anonymity, ProcessReplyClosure::anonymity_level, call_continuation(), data, ProcessReplyClosure::data, ProcessReplyClosure::eval, expiration, ProcessReplyClosure::expiration, GSF_PendingRequest::first_uid, gettext_noop, GNUNET_assert, GNUNET_BLOCK_get_key(), GNUNET_BLOCK_REPLY_OK_LAST, GNUNET_BLOCK_TYPE_FS_ONDEMAND, GNUNET_break, GNUNET_DATASTORE_remove(), GNUNET_ERROR_TYPE_DEBUG, GNUNET_FS_handle_on_demand_block(), GNUNET_h2s(), GNUNET_log, GNUNET_NO, GNUNET_OK, GNUNET_SCHEDULER_add_delayed(), GNUNET_SCHEDULER_cancel(), GNUNET_STATISTICS_update(), GNUNET_TIME_absolute_get(), GNUNET_TIME_UNIT_MINUTES, GNUNET_YES, GSF_block_ctx, GSF_dsh, GSF_PRO_PRIORITY_UNLIMITED, GSF_stats, GSF_test_get_load_too_high_(), GSF_update_datastore_delay_(), GSF_PendingRequest::have_first_uid, key, GSF_PendingRequest::local_result, MAX_RESULTS, no_more_local_results(), odc_warn_delay_task(), GSF_PendingRequestData::options, ProcessReplyClosure::priority, GSF_PendingRequestData::priority, process_local_reply(), process_reply(), GSF_PendingRequest::public_data, GSF_PendingRequest::qe, GSF_PendingRequest::qe_start, replication, ProcessReplyClosure::request_found, GSF_PendingRequest::result_count, GSF_PendingRequestData::results_found, GSF_PendingRequest::seen_null, size, ProcessReplyClosure::size, start_local_query(), GSF_PendingRequestData::start_time, type, ProcessReplyClosure::type, and GSF_PendingRequest::warn_task.

Referenced by process_local_reply(), and start_local_query().

Here is the call graph for this function:

Here is the caller graph for this function:

start_local_query()

static void start_local_query ( struct GSF_PendingRequest *  pr, uint64_t  next_uid, bool  random  )

static

Definition at line 1395 of file gnunet-service-fs_pr.c.

{
  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);
}

References call_continuation(), gettext_noop, GNUNET_BLOCK_TYPE_ANY, GNUNET_BLOCK_TYPE_FS_DBLOCK, GNUNET_DATASTORE_get_key(), GNUNET_ERROR_TYPE_DEBUG, GNUNET_h2s(), GNUNET_log, GNUNET_NO, GNUNET_SCHEDULER_add_delayed(), GNUNET_STATISTICS_update(), GNUNET_TIME_absolute_get(), GNUNET_TIME_UNIT_MINUTES, GSF_datastore_queue_size, GSF_dsh, GSF_PRO_PRIORITY_UNLIMITED, GSF_stats, GSF_PendingRequestData::options, process_local_reply(), GSF_PendingRequest::public_data, GSF_PendingRequest::qe, GSF_PendingRequest::qe_start, GSF_PendingRequestData::query, GSF_PendingRequestData::type, warn_delay_task(), and GSF_PendingRequest::warn_task.

Referenced by GSF_local_lookup_(), and process_local_reply().

Here is the call graph for this function:

Here is the caller graph for this function:

GSF_pending_request_test_target_()

int GSF_pending_request_test_target_	(	struct GSF_PendingRequest *	pr,
		const struct GNUNET_PeerIdentity *	target
	)

Is the given target a legitimate peer for forwarding the given request?

Parameters

pr	request
target

Returns

GNUNET_YES if this request could be forwarded to the given peer

Definition at line 1647 of file gnunet-service-fs_pr.c.

{
  struct GNUNET_PeerIdentity pi;
 
  if (0 == pr->origin_pid)
    return GNUNET_YES;
  GNUNET_PEER_resolve (pr->origin_pid, &pi);
  return (0 == memcmp (&pi, target, sizeof(struct GNUNET_PeerIdentity)))
         ? GNUNET_NO
         : GNUNET_YES;
}

References GNUNET_NO, GNUNET_PEER_resolve(), GNUNET_YES, and GSF_PendingRequest::origin_pid.

Referenced by consider_peer_for_forwarding(), and consider_request_for_forwarding().

Here is the call graph for this function:

Here is the caller graph for this function:

GSF_local_lookup_()

void GSF_local_lookup_	(	struct GSF_PendingRequest *	pr,
		GSF_LocalLookupContinuation	cont,
		void *	cont_cls
	)

Look up the request in the local datastore.

Parameters

pr	the pending request to process
cont	function to call at the end
cont_cls	closure for cont

Definition at line 1669 of file gnunet-service-fs_pr.c.

{
  GNUNET_assert (NULL == pr->gh);
  GNUNET_assert (NULL == pr->cadet_request);
  GNUNET_assert (NULL == pr->llc_cont);
  pr->llc_cont = cont;
  pr->llc_cont_cls = cont_cls;
#if INSANE_STATISTICS
  GNUNET_STATISTICS_update (GSF_stats,
                            gettext_noop ("# Datastore lookups initiated"),
                            1,
                            GNUNET_NO);
#endif
  start_local_query (pr, 0 /* next_uid */, true /* random */);
}

References GSF_PendingRequest::cadet_request, gettext_noop, GSF_PendingRequest::gh, GNUNET_assert, GNUNET_NO, GNUNET_STATISTICS_update(), GSF_stats, GSF_PendingRequest::llc_cont, GSF_PendingRequest::llc_cont_cls, and start_local_query().

Referenced by handle_client_start_search(), and handle_p2p_get().

Here is the call graph for this function:

Here is the caller graph for this function:

handle_p2p_put()

void handle_p2p_put	(	void *	cls,
		const struct PutMessage *	put
	)

Handle P2P "CONTENT" message.

Checks that the message is well-formed and then checks if there are any pending requests for this content and possibly passes it on (to local clients or other peers). Does NOT perform migration (content caching at this peer).

Parameters

cls	the other peer involved
put	the actual message

Definition at line 1698 of file gnunet-service-fs_pr.c.

{
  struct GSF_ConnectedPeer *cp = cls;
  uint16_t msize;
  size_t dsize;
  enum GNUNET_BLOCK_Type type;
  struct GNUNET_TIME_Absolute expiration;
  struct GNUNET_HashCode query;
  struct ProcessReplyClosure prq;
  struct GNUNET_TIME_Relative block_time;
  double putl;
  struct PutMigrationContext *pmc;
 
  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 (GNUNET_OK !=
      GNUNET_BLOCK_check_block (GSF_block_ctx,
                                type,
                                &put[1],
                                dsize))
  {
    GNUNET_break_op (0);
    return;
  }
  if (GNUNET_OK !=
      GNUNET_BLOCK_get_key (GSF_block_ctx,
                            type,
                            &put[1],
                            dsize,
                            &query))
  {
    GNUNET_break_op (0);
    return;
  }
  GNUNET_STATISTICS_update (GSF_stats,
                            gettext_noop ("# GAP PUT messages received"),
                            1,
                            GNUNET_NO);
  /* now, lookup 'query' */
  prq.data = (const void *) &put[1];
  prq.sender = cp;
  prq.size = dsize;
  prq.type = type;
  prq.expiration = expiration;
  prq.priority = 0;
  prq.anonymity_level = UINT32_MAX;
  prq.request_found = GNUNET_NO;
  GNUNET_CONTAINER_multihashmap_get_multiple (pr_map,
                                              &query,
                                              &process_reply,
                                              &prq);
  if (NULL != cp)
  {
    GSF_connected_peer_change_preference_ (cp,
                                           CONTENT_BANDWIDTH_VALUE
                                           + 1000 * 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,
                "Replicating result for query `%s' with priority %u\n",
                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 */,
                                      0 /* replication */,
                                      expiration,
                                      1 + prq.priority,
                                      MAX_DATASTORE_QUEUE,
                                      &put_migration_continuation,
                                      pmc))
    {
      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,
                test_put_load_too_high (prq.priority));
  }
  putl = GNUNET_LOAD_get_load (datastore_put_load);
  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);
    block_time = GNUNET_TIME_relative_multiply (
      GNUNET_TIME_UNIT_MILLISECONDS,
      5000 + GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
                                       (unsigned int) (60000 * putl * putl)));
    GNUNET_log (
      GNUNET_ERROR_TYPE_DEBUG,
      "Asking to stop migration for %s because of load %f and events %d/%d\n",
      GNUNET_STRINGS_relative_time_to_string (block_time, GNUNET_YES),
      putl,
      active_to_migration,
      (GNUNET_NO == prq.request_found));
    GSF_block_peer_migration_ (cp,
                               GNUNET_TIME_relative_to_absolute (block_time));
  }
}

References active_to_migration, ProcessReplyClosure::anonymity_level, CONTENT_BANDWIDTH_VALUE, ProcessReplyClosure::data, datastore_put_load, expiration, PutMessage::expiration, ProcessReplyClosure::expiration, gettext_noop, GNUNET_assert, GNUNET_BLOCK_check_block(), GNUNET_BLOCK_get_key(), GNUNET_break_op, GNUNET_CONTAINER_multihashmap_get_multiple(), GNUNET_CRYPTO_QUALITY_WEAK, GNUNET_CRYPTO_random_u32(), GNUNET_DATASTORE_put(), GNUNET_ERROR_TYPE_DEBUG, GNUNET_h2s(), GNUNET_i2s(), GNUNET_LOAD_get_load(), GNUNET_log, GNUNET_new, GNUNET_NO, GNUNET_OK, GNUNET_PEER_resolve(), GNUNET_STATISTICS_update(), GNUNET_STRINGS_relative_time_to_string(), GNUNET_SYSERR, GNUNET_TIME_absolute_get(), GNUNET_TIME_absolute_min(), GNUNET_TIME_absolute_ntoh(), GNUNET_TIME_relative_multiply(), GNUNET_TIME_relative_to_absolute(), GNUNET_TIME_UNIT_MILLISECONDS, GNUNET_TIME_UNIT_YEARS, GNUNET_TIME_UNIT_ZERO_ABS, GNUNET_YES, GSF_block_ctx, GSF_block_peer_migration_(), GSF_connected_peer_change_preference_(), GSF_cover_content_count, GSF_dsh, GSF_get_peer_performance_data_(), GSF_stats, PutMessage::header, MAX_DATASTORE_QUEUE, PutMigrationContext::origin, pid, pr_map, ProcessReplyClosure::priority, process_reply(), put_migration_continuation(), ProcessReplyClosure::request_found, PutMigrationContext::requested, GSF_PeerPerformanceData::respect, ProcessReplyClosure::sender, GNUNET_MessageHeader::size, ProcessReplyClosure::size, PutMigrationContext::start, test_put_load_too_high(), type, PutMessage::type, and ProcessReplyClosure::type.

Here is the call graph for this function:

GSF_pending_request_test_active_()

int GSF_pending_request_test_active_

(

struct GSF_PendingRequest *

pr

)

Check if the given request is still active.

Parameters

pr	pending request

Returns

GNUNET_YES if the request is still active

Definition at line 1840 of file gnunet-service-fs_pr.c.

{
  return (NULL != pr->rh) ? GNUNET_YES : GNUNET_NO;
}

References GNUNET_NO, GNUNET_YES, and GSF_PendingRequest::rh.

Referenced by consider_peer_for_forwarding(), get_latest(), GSF_consider_forwarding(), GSF_plan_add_(), and merge_pr().

Here is the caller graph for this function:

GSF_pending_request_init_()

void GSF_pending_request_init_

(

void

)

Setup the subsystem.

Definition at line 1850 of file gnunet-service-fs_pr.c.

{
  if (GNUNET_OK !=
      GNUNET_CONFIGURATION_get_value_number (GSF_cfg,
                                             "fs",
                                             "MAX_PENDING_REQUESTS",
                                             &max_pending_requests))
  {
    GNUNET_log_config_missing (GNUNET_ERROR_TYPE_INFO,
                               "fs",
                               "MAX_PENDING_REQUESTS");
  }
  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, GNUNET_YES);
  requests_by_expiration_heap =
    GNUNET_CONTAINER_heap_create (GNUNET_CONTAINER_HEAP_ORDER_MIN);
}

References active_to_migration, DATASTORE_LOAD_AUTODECLINE, datastore_put_load, GNUNET_CONFIGURATION_get_value_number(), GNUNET_CONFIGURATION_get_value_yesno(), GNUNET_CONTAINER_heap_create(), GNUNET_CONTAINER_HEAP_ORDER_MIN, GNUNET_CONTAINER_multihashmap_create(), GNUNET_ERROR_TYPE_INFO, GNUNET_LOAD_value_init(), GNUNET_log_config_missing(), GNUNET_OK, GNUNET_YES, GSF_cfg, max_pending_requests, pr_map, and requests_by_expiration_heap.

Referenced by run().

Here is the call graph for this function:

Here is the caller graph for this function:

GSF_pending_request_done_()

void GSF_pending_request_done_

(

void

)

Shutdown the subsystem.

Definition at line 1875 of file gnunet-service-fs_pr.c.

{
  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);
  requests_by_expiration_heap = NULL;
  GNUNET_LOAD_value_free (datastore_put_load);
  datastore_put_load = NULL;
}

References clean_request(), datastore_put_load, GNUNET_CONTAINER_heap_destroy(), GNUNET_CONTAINER_multihashmap_destroy(), GNUNET_CONTAINER_multihashmap_iterate(), GNUNET_LOAD_value_free, pr_map, and requests_by_expiration_heap.

Referenced by shutdown_task().

Here is the call graph for this function:

Here is the caller graph for this function:

Variable Documentation

pr_map

struct GNUNET_CONTAINER_MultiHashMap* pr_map

static

All pending requests, ordered by the query.

Entries are of type 'struct GSF_PendingRequest*'.

Definition at line 205 of file gnunet-service-fs_pr.c.

Referenced by clean_request(), GSF_iterate_pending_requests_(), GSF_pending_request_cancel_(), GSF_pending_request_create_(), GSF_pending_request_done_(), GSF_pending_request_init_(), and handle_p2p_put().

datastore_put_load

struct GNUNET_LOAD_Value* datastore_put_load

static

Datastore 'PUT' load tracking.

Definition at line 211 of file gnunet-service-fs_pr.c.

Referenced by GSF_pending_request_done_(), GSF_pending_request_init_(), handle_p2p_put(), put_migration_continuation(), and test_put_load_too_high().

active_to_migration

int active_to_migration

static

Are we allowed to migrate content to this peer.

Definition at line 217 of file gnunet-service-fs_pr.c.

Referenced by GSF_pending_request_init_(), handle_dht_reply(), and handle_p2p_put().

requests_by_expiration_heap

struct GNUNET_CONTAINER_Heap* requests_by_expiration_heap

static

Heap with the request that will expire next at the top.

Contains pointers of type "struct PendingRequest*"; these will also be aliased from the "requests_by_peer" data structures and the "requests_by_query" table. Note that requests from our clients don't expire and are thus NOT in the "requests_by_expiration" (or the "requests_by_peer" tables).

Definition at line 228 of file gnunet-service-fs_pr.c.

Referenced by GSF_pending_request_create_(), GSF_pending_request_done_(), and GSF_pending_request_init_().

max_pending_requests

unsigned long long max_pending_requests = (32 * 1024)

static

Maximum number of requests (from other peers, overall) that we're willing to have pending at any given point in time.

Can be changed via the configuration file (32k is just the default).

Definition at line 236 of file gnunet-service-fs_pr.c.

Referenced by GSF_pending_request_create_(), and GSF_pending_request_init_().