gnunet-service-fs_cp.c File Reference

API to handle 'connected peers'. More...

#include "gnunet_common.h"
#include "gnunet_pils_service.h"
#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_pe.h"
#include "gnunet-service-fs_pr.h"
#include "gnunet-service-fs_push.h"
#include "gnunet_peerstore_service.h"

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

Go to the source code of this file.

Data Structures
struct	GSF_PeerTransmitHandle
	Handle to cancel a transmission request. More...
struct	GSF_DelayedHandle
	Handle for an entry in our delay list. More...
struct	PeerRequest
	Information per peer and request. More...
struct	GSF_ConnectedPeer
	A connected peer. More...
struct	TestExistClosure
	Closure for test_exist_cb(). More...
struct	IterationContext
	Closure for call_iterator(). More...

Macros
#define	RUNAVG_DELAY_N   16
	Ratio for moving average delay calculation.
#define	RESPECT_FLUSH_FREQ
	How often do we flush respect values to disk?
#define	REPLY_TIMEOUT
	After how long do we discard a reply?
#define	INSANE_STATISTICS   GNUNET_NO
	Collect an instance number of statistics? May cause excessive IPC.
#define	N   ((double) 128.0)

Functions
void	GSF_update_peer_latency_ (const struct GNUNET_PeerIdentity *id, struct GNUNET_TIME_Relative latency)
	Update the latency information kept for the given peer.
struct GSF_PeerPerformanceData *	GSF_get_peer_performance_data_ (struct GSF_ConnectedPeer *cp)
	Return the performance data record for the given peer.
static void	peer_transmit (struct GSF_ConnectedPeer *cp)
	Core is ready to transmit to a peer, get the message.
static void	schedule_transmission (struct GSF_PeerTransmitHandle *pth)
	If ready (bandwidth reserved), try to schedule transmission via core for the given handle.
static void	peer_respect_cb (void *cls, const struct GNUNET_PEERSTORE_Record *record, const char *emsg)
	Function called by PEERSTORE with peer respect record.
static int	consider_peer_for_forwarding (void *cls, const struct GNUNET_HashCode *key, struct GSF_PendingRequest *pr)
	Function called for each pending request whenever a new peer connects, giving us a chance to decide about submitting the existing request to the new peer.
void *	GSF_peer_connect_handler (void *cls, const struct GNUNET_PeerIdentity *peer, struct GNUNET_MQ_Handle *mq, enum GNUNET_CORE_PeerClass class)
	A peer connected to us.
static void	revive_migration (void *cls)
	It may be time to re-start migrating content to this peer.
struct GSF_ConnectedPeer *	GSF_peer_get_ (const struct GNUNET_PeerIdentity *peer)
	Get a handle for a connected peer.
void	handle_p2p_migration_stop (void *cls, const struct MigrationStopMessage *msm)
	Handle P2P GNUNET_MESSAGE_TYPE_FS_MIGRATION_STOP message.
static void	free_pending_request (struct PeerRequest *peerreq)
	Free resources associated with the given peer request.
static int	cancel_pending_request (void *cls, const struct GNUNET_HashCode *query, void *value)
	Cancel all requests associated with the peer.
static void	peer_request_destroy (void *cls)
	Free the given request.
static void	transmit_delayed_now (void *cls)
	The artificial delay is over, transmit the message now.
static struct GNUNET_TIME_Relative	get_randomized_delay ()
	Get the randomized delay a response should be subjected to.
static void	handle_p2p_reply (void *cls, enum GNUNET_BLOCK_ReplyEvaluationResult eval, struct GSF_PendingRequest *pr, uint32_t reply_anonymity_level, struct GNUNET_TIME_Absolute expiration, struct GNUNET_TIME_Absolute last_transmission, enum GNUNET_BLOCK_Type type, const void *data, size_t data_len)
	Handle a reply to a pending request.
static int	change_peer_respect (struct GSF_ConnectedPeer *cp, int value)
	Increase the peer's respect by a value.
static int32_t	bound_priority (uint32_t prio_in, struct GSF_ConnectedPeer *cp)
	We've received a request with the specified priority.
static int32_t	bound_ttl (int32_t ttl_in, uint32_t prio)
	The priority level imposes a bound on the maximum value for the ttl that can be requested.
static int	test_exist_cb (void *cls, const struct GNUNET_HashCode *hc, void *value)
	Test if the query already exists.
void	handle_p2p_get (void *cls, const struct GetMessage *gm)
	Handle P2P "QUERY" message.
void	GSF_peer_transmit_ (struct GSF_ConnectedPeer *cp, int is_query, uint32_t priority, struct GNUNET_MQ_Envelope *env)
	Transmit a message to the given peer as soon as possible.
void	GSF_peer_update_performance_ (struct GSF_ConnectedPeer *cp, struct GNUNET_TIME_Absolute request_time, uint32_t request_priority)
	Report on receiving a reply; update the performance record of the given peer.
void	GSF_peer_update_responder_client_ (struct GSF_ConnectedPeer *cp, struct GSF_LocalClient *initiator_client)
	Report on receiving a reply in response to an initiating client.
void	GSF_peer_update_responder_peer_ (struct GSF_ConnectedPeer *cp, const struct GSF_ConnectedPeer *initiator_peer)
	Report on receiving a reply in response to an initiating peer.
static int	flush_respect (void *cls, const struct GNUNET_PeerIdentity *key, void *value)
	Write peer-respect information to a file - flush the buffer entry!
void	GSF_peer_disconnect_handler (void *cls, const struct GNUNET_PeerIdentity *peer, void *internal_cls)
	A peer disconnected from us.
static int	call_iterator (void *cls, const struct GNUNET_PeerIdentity *key, void *value)
	Function that calls the callback for each peer.
void	GSF_iterate_connected_peers_ (GSF_ConnectedPeerIterator it, void *it_cls)
	Iterate over all connected peers.
void	GSF_connected_peer_get_identity_ (const struct GSF_ConnectedPeer *cp, struct GNUNET_PeerIdentity *id)
	Obtain the identity of a connected peer.
const struct GNUNET_PeerIdentity *	GSF_connected_peer_get_identity2_ (const struct GSF_ConnectedPeer *cp)
	Obtain the identity of a connected peer.
void	GSF_block_peer_migration_ (struct GSF_ConnectedPeer *cp, struct GNUNET_TIME_Absolute block_time)
	Ask a peer to stop migrating data to us until the given point in time.
void	GSF_connected_peer_change_preference_ (struct GSF_ConnectedPeer *cp, uint64_t pref)
	Notify core about a preference we have for the given peer (to allocate more resources towards it).
static void	cron_flush_respect (void *cls)
	Call this method periodically to flush respect information to disk.
void	GSF_connected_peer_init_ ()
	Initialize peer management subsystem.
void	GSF_connected_peer_done_ ()
	Shutdown peer management subsystem.
static int	clean_local_client (void *cls, const struct GNUNET_PeerIdentity *key, void *value)
	Iterator to remove references to LC entry.
void	GSF_handle_local_client_disconnect_ (const struct GSF_LocalClient *lc)
	Notification that a local client disconnected.

Variables
static struct GNUNET_CONTAINER_MultiPeerMap *	cp_map
	Map from peer identities to struct GSF_ConnectPeer entries.
static struct GNUNET_PEERSTORE_Handle *	peerstore
	Handle to peerstore service.
static struct GNUNET_SCHEDULER_Task *	fr_task
	Task used to flush respect values to disk.

Detailed Description

API to handle 'connected peers'.

Author

Christian Grothoff

Definition in file gnunet-service-fs_cp.c.

Macro Definition Documentation

RUNAVG_DELAY_N

#define RUNAVG_DELAY_N   16

Ratio for moving average delay calculation.

The previous average goes in with a factor of (n-1) into the calculation. Must be > 0.

Definition at line 43 of file gnunet-service-fs_cp.c.

RESPECT_FLUSH_FREQ

#define RESPECT_FLUSH_FREQ

Value:

          GNUNET_TIME_relative_multiply ( \
          GNUNET_TIME_UNIT_MINUTES, 5)

How often do we flush respect values to disk?

Definition at line 48 of file gnunet-service-fs_cp.c.

{
  struct GSF_PeerTransmitHandle *next;
 
  struct GSF_PeerTransmitHandle *prev;
 
  struct GNUNET_TIME_Absolute transmission_request_start_time;
 
  struct GNUNET_MQ_Envelope *env;
 
  struct GSF_ConnectedPeer *cp;
 
  int is_query;
 
  uint32_t priority;
};
 
 
struct GSF_DelayedHandle
{
  struct GSF_DelayedHandle *next;
 
  struct GSF_DelayedHandle *prev;
 
  struct GSF_ConnectedPeer *cp;
 
  struct GNUNET_MQ_Envelope *env;
 
  struct GNUNET_SCHEDULER_Task *delay_task;
 
  size_t msize;
};
 
 
struct PeerRequest
{
  struct GSF_PendingRequest *pr;
 
  struct GSF_ConnectedPeer *cp;
 
  struct GNUNET_SCHEDULER_Task *kill_task;
};
 
 
struct GSF_ConnectedPeer
{
  struct GSF_PeerPerformanceData ppd;
 
  struct GNUNET_TIME_Absolute last_migration_block;
 
  struct GNUNET_SCHEDULER_Task *mig_revive_task;
 
  struct GSF_PeerTransmitHandle *pth_head;
 
  struct GSF_PeerTransmitHandle *pth_tail;
 
  struct GSF_DelayedHandle *delayed_head;
 
  struct GSF_DelayedHandle *delayed_tail;
 
  struct GNUNET_SCHEDULER_Task *rc_delay_task;
 
  struct GNUNET_CONTAINER_MultiHashMap *request_map;
 
  struct GNUNET_MQ_Handle *mq;
 
  uint64_t inc_preference;
 
  unsigned int delay_queue_size;
 
  uint32_t disk_respect;
 
  unsigned int last_p2p_replies_woff;
 
  unsigned int last_client_replies_woff;
 
  unsigned int last_request_times_off;
 
  struct GNUNET_PEERSTORE_IterateContext *respect_iterate_req;
};
 
 
static struct GNUNET_CONTAINER_MultiPeerMap *cp_map;
 
static struct GNUNET_PEERSTORE_Handle *peerstore;
 
static struct GNUNET_SCHEDULER_Task *fr_task;
 
 
void
GSF_update_peer_latency_ (const struct GNUNET_PeerIdentity *id,
                          struct GNUNET_TIME_Relative latency)
{
  struct GSF_ConnectedPeer *cp;
 
  cp = GSF_peer_get_ (id);
  if (NULL == cp)
    return; /* we're not yet connected at the core level, ignore */
  GNUNET_LOAD_value_set_decline (cp->ppd.transmission_delay,
                                 latency);
}
 
 
struct GSF_PeerPerformanceData *
GSF_get_peer_performance_data_ (struct GSF_ConnectedPeer *cp)
{
  return &cp->ppd;
}
 
 
static void
peer_transmit (struct GSF_ConnectedPeer *cp);
 
 
static void
schedule_transmission (struct GSF_PeerTransmitHandle *pth)
{
  struct GSF_ConnectedPeer *cp;
  struct GNUNET_PeerIdentity target;
 
  cp = pth->cp;
  GNUNET_assert (0 != cp->ppd.pid);
  GNUNET_PEER_resolve (cp->ppd.pid, &target);
 
  peer_transmit (cp);
}
 
 
static void
peer_transmit (struct GSF_ConnectedPeer *cp)
{
  struct GSF_PeerTransmitHandle *pth = cp->pth_head;
  struct GSF_PeerTransmitHandle *pos;
 
  if (NULL == pth)
    return;
  GNUNET_CONTAINER_DLL_remove (cp->pth_head,
                               cp->pth_tail,
                               pth);
  if (GNUNET_YES == pth->is_query)
  {
    cp->ppd.last_request_times[(cp->last_request_times_off++)
                               % MAX_QUEUE_PER_PEER] =
      GNUNET_TIME_absolute_get ();
    GNUNET_assert (0 < cp->ppd.pending_queries--);
  }
  else if (GNUNET_NO == pth->is_query)
  {
    GNUNET_assert (0 < cp->ppd.pending_replies--);
  }
  GNUNET_LOAD_update (cp->ppd.transmission_delay,
                      GNUNET_TIME_absolute_get_duration
                        (pth->transmission_request_start_time).rel_value_us);
  GNUNET_MQ_send (cp->mq,
                  pth->env);
  GNUNET_free (pth);
  if (NULL != (pos = cp->pth_head))
  {
    GNUNET_assert (pos != pth);
    schedule_transmission (pos);
  }
}
 
 
static void
peer_respect_cb (void *cls,
                 const struct GNUNET_PEERSTORE_Record *record,
                 const char *emsg)
{
  struct GSF_ConnectedPeer *cp = cls;
 
  GNUNET_assert (NULL != cp->respect_iterate_req);
  if ((NULL == record) && (NULL == emsg))
  {
    cp->respect_iterate_req = NULL;
    return;
  }
  if ((NULL != record) &&
      (sizeof(cp->disk_respect) == record->value_size))
  {
    cp->disk_respect = *((uint32_t *) record->value);
    cp->ppd.respect += *((uint32_t *) record->value);
  }
  GSF_push_start_ (cp);
  if (NULL != record)
  {
    GNUNET_PEERSTORE_iteration_stop (cp->respect_iterate_req);
    cp->respect_iterate_req = NULL;
    return;
  }
  GNUNET_PEERSTORE_iteration_next (cp->respect_iterate_req, 1);
}
 
 
static int
consider_peer_for_forwarding (void *cls,
                              const struct GNUNET_HashCode *key,
                              struct GSF_PendingRequest *pr)
{
  struct GSF_ConnectedPeer *cp = cls;
  struct GNUNET_PeerIdentity pid;
 
  if (GNUNET_YES !=
      GSF_pending_request_test_active_ (pr))
    return GNUNET_YES; /* request is not actually active, skip! */
  GSF_connected_peer_get_identity_ (cp, &pid);
  if (GNUNET_YES !=
      GSF_pending_request_test_target_ (pr, &pid))
  {
    GNUNET_STATISTICS_update (GSF_stats,
                              gettext_noop ("# Loopback routes suppressed"),
                              1,
                              GNUNET_NO);
    return GNUNET_YES;
  }
  GSF_plan_add_ (cp, pr);
  return GNUNET_YES;
}
 
 
void *
GSF_peer_connect_handler (void *cls,
                          const struct GNUNET_PeerIdentity *peer,
                          struct GNUNET_MQ_Handle *mq,
                          enum GNUNET_CORE_PeerClass class)
{
  const struct GNUNET_PeerIdentity *my_identity;
  struct GSF_ConnectedPeer *cp;
 
  my_identity = GNUNET_PILS_get_identity (GSF_pils);
  GNUNET_assert (my_identity);
 
  if (0 == GNUNET_memcmp (my_identity, peer))
    return NULL;
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Connected to peer %s\n",
              GNUNET_i2s (peer));
  cp = GNUNET_new (struct GSF_ConnectedPeer);
  cp->ppd.pid = GNUNET_PEER_intern (peer);
  cp->ppd.peer = peer;
  cp->mq = mq;
  cp->ppd.transmission_delay = GNUNET_LOAD_value_init (GNUNET_TIME_UNIT_ZERO);
 
  cp->request_map = GNUNET_CONTAINER_multihashmap_create (128,
                                                          GNUNET_YES);
  GNUNET_break (GNUNET_OK ==
                GNUNET_CONTAINER_multipeermap_put (cp_map,
                                                   GSF_connected_peer_get_identity2_
                                                   (
                                                     cp),
                                                   cp,
                                                   GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
  GNUNET_STATISTICS_set (GSF_stats,
                         gettext_noop ("# peers connected"),
                         GNUNET_CONTAINER_multipeermap_size (cp_map),
                         GNUNET_NO);
  cp->respect_iterate_req
    = GNUNET_PEERSTORE_iteration_start (peerstore,
                                        "fs",
                                        peer,
                                        "respect",
                                        &peer_respect_cb,
                                        cp);
  GSF_iterate_pending_requests_ (&consider_peer_for_forwarding,
                                 cp);
  return cp;
}
 
 
static void
revive_migration (void *cls)
{
  struct GSF_ConnectedPeer *cp = cls;
  struct GNUNET_TIME_Relative bt;
 
  cp->mig_revive_task = NULL;
  bt = GNUNET_TIME_absolute_get_remaining (cp->ppd.migration_blocked_until);
  if (0 != bt.rel_value_us)
  {
    /* still time left... */
    cp->mig_revive_task =
      GNUNET_SCHEDULER_add_delayed (bt, &revive_migration, cp);
    return;
  }
  GSF_push_start_ (cp);
}
 
 
struct GSF_ConnectedPeer *
GSF_peer_get_ (const struct GNUNET_PeerIdentity *peer)
{
  if (NULL == cp_map)
    return NULL;
  return GNUNET_CONTAINER_multipeermap_get (cp_map, peer);
}
 
 
void
handle_p2p_migration_stop (void *cls,
                           const struct MigrationStopMessage *msm)
{
  struct GSF_ConnectedPeer *cp = cls;
  struct GNUNET_TIME_Relative bt;
 
  GNUNET_STATISTICS_update (GSF_stats,
                            gettext_noop ("# migration stop messages received"),
                            1, GNUNET_NO);
  bt = GNUNET_TIME_relative_ntoh (msm->duration);
  GNUNET_log (GNUNET_ERROR_TYPE_INFO,
              _ ("Migration of content to peer `%s' blocked for %s\n"),
              GNUNET_i2s (cp->ppd.peer),
              GNUNET_STRINGS_relative_time_to_string (bt, GNUNET_YES));
  cp->ppd.migration_blocked_until = GNUNET_TIME_relative_to_absolute (bt);
  if ((NULL == cp->mig_revive_task) &&
      (NULL == cp->respect_iterate_req))
  {
    GSF_push_stop_ (cp);
    cp->mig_revive_task =
      GNUNET_SCHEDULER_add_delayed (bt,
                                    &revive_migration, cp);
  }
}
 
 
static void
free_pending_request (struct PeerRequest *peerreq)
{
  struct GSF_ConnectedPeer *cp = peerreq->cp;
  struct GSF_PendingRequestData *prd;
 
  prd = GSF_pending_request_get_data_ (peerreq->pr);
  if (NULL != peerreq->kill_task)
  {
    GNUNET_SCHEDULER_cancel (peerreq->kill_task);
    peerreq->kill_task = NULL;
  }
  GNUNET_STATISTICS_update (GSF_stats,
                            gettext_noop ("# P2P searches active"),
                            -1,
                            GNUNET_NO);
  GNUNET_break (GNUNET_YES ==
                GNUNET_CONTAINER_multihashmap_remove (cp->request_map,
                                                      &prd->query,
                                                      peerreq));
  GNUNET_free (peerreq);
}
 
 
static int
cancel_pending_request (void *cls,
                        const struct GNUNET_HashCode *query,
                        void *value)
{
  struct PeerRequest *peerreq = value;
  struct GSF_PendingRequest *pr = peerreq->pr;
 
  free_pending_request (peerreq);
  GSF_pending_request_cancel_ (pr,
                               GNUNET_NO);
  return GNUNET_OK;
}
 
 
static void
peer_request_destroy (void *cls)
{
  struct PeerRequest *peerreq = cls;
  struct GSF_PendingRequest *pr = peerreq->pr;
  struct GSF_PendingRequestData *prd;
 
  peerreq->kill_task = NULL;
  prd = GSF_pending_request_get_data_ (pr);
  cancel_pending_request (NULL,
                          &prd->query,
                          peerreq);
}
 
 
static void
transmit_delayed_now (void *cls)
{
  struct GSF_DelayedHandle *dh = cls;
  struct GSF_ConnectedPeer *cp = dh->cp;
 
  GNUNET_CONTAINER_DLL_remove (cp->delayed_head,
                               cp->delayed_tail,
                               dh);
  cp->delay_queue_size--;
  GSF_peer_transmit_ (cp,
                      GNUNET_NO,
                      UINT32_MAX,
                      dh->env);
  GNUNET_free (dh);
}
 
 
static struct GNUNET_TIME_Relative
get_randomized_delay ()
{
  struct GNUNET_TIME_Relative ret;
 
  ret =
    GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MILLISECONDS,
                                   GNUNET_CRYPTO_random_u32
                                     (GNUNET_CRYPTO_QUALITY_WEAK,
                                     2 * GSF_avg_latency.rel_value_us + 1));
#if INSANE_STATISTICS
  GNUNET_STATISTICS_update (GSF_stats,
                            gettext_noop
                              ("# artificial delays introduced (ms)"),
                            ret.rel_value_us / 1000LL, GNUNET_NO);
#endif
  return ret;
}
 
 
static void
handle_p2p_reply (void *cls,
                  enum GNUNET_BLOCK_ReplyEvaluationResult eval,
                  struct GSF_PendingRequest *pr,
                  uint32_t reply_anonymity_level,
                  struct GNUNET_TIME_Absolute expiration,
                  struct GNUNET_TIME_Absolute last_transmission,
                  enum GNUNET_BLOCK_Type type,
                  const void *data,
                  size_t data_len)
{
  struct PeerRequest *peerreq = cls;
  struct GSF_ConnectedPeer *cp = peerreq->cp;
  struct GSF_PendingRequestData *prd;
  struct GNUNET_MQ_Envelope *env;
  struct PutMessage *pm;
  size_t msize;
 
  GNUNET_assert (data_len + sizeof(struct PutMessage) <
                 GNUNET_MAX_MESSAGE_SIZE);
  GNUNET_assert (peerreq->pr == pr);
  prd = GSF_pending_request_get_data_ (pr);
  if (NULL == data)
  {
    free_pending_request (peerreq);
    return;
  }
  GNUNET_break (GNUNET_BLOCK_TYPE_ANY != type);
  if ( (prd->type != type) &&
       (GNUNET_BLOCK_TYPE_ANY != prd->type) )
  {
    GNUNET_STATISTICS_update (GSF_stats,
                              "# replies dropped due to type mismatch",
                              1, GNUNET_NO);
    return;
  }
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Transmitting result for query `%s' to peer\n",
              GNUNET_h2s (&prd->query));
  GNUNET_STATISTICS_update (GSF_stats,
                            "# replies received for other peers",
                            1,
                            GNUNET_NO);
  msize = sizeof(struct PutMessage) + data_len;
  if (msize >= GNUNET_MAX_MESSAGE_SIZE)
  {
    GNUNET_break (0);
    return;
  }
  if ( (UINT32_MAX != reply_anonymity_level) &&
       (reply_anonymity_level > 1) )
  {
    if (reply_anonymity_level - 1 > GSF_cover_content_count)
    {
      GNUNET_STATISTICS_update (GSF_stats,
                                "# replies dropped due to insufficient cover traffic",
                                1, GNUNET_NO);
      return;
    }
    GSF_cover_content_count -= (reply_anonymity_level - 1);
  }
 
  env = GNUNET_MQ_msg_extra (pm,
                             data_len,
                             GNUNET_MESSAGE_TYPE_FS_PUT);
  pm->type = htonl (type);
  pm->expiration = GNUNET_TIME_absolute_hton (expiration);
  GNUNET_memcpy (&pm[1],
                 data,
                 data_len);
  if ((UINT32_MAX != reply_anonymity_level) &&
      (0 != reply_anonymity_level) &&
      (GNUNET_YES == GSF_enable_randomized_delays))
  {
    struct GSF_DelayedHandle *dh;
 
    dh = GNUNET_new (struct GSF_DelayedHandle);
    dh->cp = cp;
    dh->env = env;
    dh->msize = msize;
    GNUNET_CONTAINER_DLL_insert (cp->delayed_head,
                                 cp->delayed_tail,
                                 dh);
    cp->delay_queue_size++;
    dh->delay_task =
      GNUNET_SCHEDULER_add_delayed (get_randomized_delay (),
                                    &transmit_delayed_now,
                                    dh);
  }
  else
  {
    GSF_peer_transmit_ (cp,
                        GNUNET_NO,
                        UINT32_MAX,
                        env);
  }
  if (GNUNET_BLOCK_REPLY_OK_LAST != eval)
    return;
  if (NULL == peerreq->kill_task)
  {
    GNUNET_STATISTICS_update (GSF_stats,
                              "# P2P searches destroyed due to ultimate reply",
                              1,
                              GNUNET_NO);
    peerreq->kill_task =
      GNUNET_SCHEDULER_add_now (&peer_request_destroy,
                                peerreq);
  }
}
 
 
static int
change_peer_respect (struct GSF_ConnectedPeer *cp, int value)
{
  if (0 == value)
    return 0;
  GNUNET_assert (NULL != cp);
  if (value > 0)
  {
    if (cp->ppd.respect + value < cp->ppd.respect)
    {
      value = UINT32_MAX - cp->ppd.respect;
      cp->ppd.respect = UINT32_MAX;
    }
    else
      cp->ppd.respect += value;
  }
  else
  {
    if (cp->ppd.respect < -value)
    {
      value = -cp->ppd.respect;
      cp->ppd.respect = 0;
    }
    else
      cp->ppd.respect += value;
  }
  return value;
}
 
 
static int32_t
bound_priority (uint32_t prio_in,
                struct GSF_ConnectedPeer *cp)
{
#define N ((double) 128.0)
  uint32_t ret;
  double rret;
  int ld;
 
  ld = GSF_test_get_load_too_high_ (0);
  if (GNUNET_SYSERR == ld)
  {
#if INSANE_STATISTICS
    GNUNET_STATISTICS_update (GSF_stats,
                              gettext_noop
                                ("# requests done for free (low load)"), 1,
                              GNUNET_NO);
#endif
    return 0;                   /* excess resources */
  }
  if (prio_in > INT32_MAX)
    prio_in = INT32_MAX;
  ret = -change_peer_respect (cp, -(int) prio_in);
  if (ret > 0)
  {
    if (ret > GSF_current_priorities + N)
      rret = GSF_current_priorities + N;
    else
      rret = ret;
    GSF_current_priorities = (GSF_current_priorities * (N - 1) + rret) / N;
  }
  if ((GNUNET_YES == ld) && (ret > 0))
  {
    /* try with charging */
    ld = GSF_test_get_load_too_high_ (ret);
  }
  if (GNUNET_YES == ld)
  {
    GNUNET_STATISTICS_update (GSF_stats,
                              gettext_noop
                                ("# request dropped, priority insufficient"), 1,
                              GNUNET_NO);
    /* undo charge */
    change_peer_respect (cp, (int) ret);
    return -1;                  /* not enough resources */
  }
  else
  {
    GNUNET_STATISTICS_update (GSF_stats,
                              gettext_noop
                                ("# requests done for a price (normal load)"),
                              1,
                              GNUNET_NO);
  }
#undef N
  return ret;
}
 
 
static int32_t
bound_ttl (int32_t ttl_in,
           uint32_t prio)
{
  unsigned long long allowed;
 
  if (ttl_in <= 0)
    return ttl_in;
  allowed = ((unsigned long long) prio) * TTL_DECREMENT / 1000;
  if (ttl_in > allowed)
  {
    if (allowed >= (1 << 30))
      return 1 << 30;
    return allowed;
  }
  return ttl_in;
}
 
 
struct TestExistClosure
{
  int32_t priority;
 
  int32_t ttl;
 
  enum GNUNET_BLOCK_Type type;
 
  int finished;
};
 
 
static int
test_exist_cb (void *cls,
               const struct GNUNET_HashCode *hc,
               void *value)
{
  struct TestExistClosure *tec = cls;
  struct PeerRequest *peerreq = value;
  struct GSF_PendingRequest *pr;
  struct GSF_PendingRequestData *prd;
 
  pr = peerreq->pr;
  prd = GSF_pending_request_get_data_ (pr);
  if (prd->type != tec->type)
    return GNUNET_YES;
  if (prd->ttl.abs_value_us >=
      GNUNET_TIME_absolute_get ().abs_value_us + tec->ttl * 1000LL)
  {
    /* existing request has higher TTL, drop new one! */
    prd->priority += tec->priority;
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Have existing request with higher TTL, dropping new request.\n");
    GNUNET_STATISTICS_update (GSF_stats,
                              gettext_noop
                                ("# requests dropped due to higher-TTL request")
                              ,
                              1, GNUNET_NO);
    tec->finished = GNUNET_YES;
    return GNUNET_NO;
  }
  /* existing request has lower TTL, drop old one! */
  tec->priority += prd->priority;
  free_pending_request (peerreq);
  GSF_pending_request_cancel_ (pr,
                               GNUNET_YES);
  return GNUNET_NO;
}
 
 
void
handle_p2p_get (void *cls,
                const struct GetMessage *gm)
{
  struct GSF_ConnectedPeer *cps = cls;
  struct PeerRequest *peerreq;
  struct GSF_PendingRequest *pr;
  struct GSF_ConnectedPeer *cp;
  const struct GNUNET_PeerIdentity *target;
  enum GSF_PendingRequestOptions options;
  uint16_t msize;
  unsigned int bits;
  const struct GNUNET_PeerIdentity *opt;
  uint32_t bm;
  size_t bfsize;
  uint32_t ttl_decrement;
  struct TestExistClosure tec;
  GNUNET_PEER_Id spid;
  const struct GSF_PendingRequestData *prd;
 
  msize = ntohs (gm->header.size);
  tec.type = ntohl (gm->type);
  bm = ntohl (gm->hash_bitmap);
  bits = 0;
  while (bm > 0)
  {
    if (1 == (bm & 1))
      bits++;
    bm >>= 1;
  }
  opt = (const struct GNUNET_PeerIdentity *) &gm[1];
  bfsize = msize - sizeof(struct GetMessage) - bits * sizeof(struct
                                                             GNUNET_PeerIdentity);
  GNUNET_STATISTICS_update (GSF_stats,
                            gettext_noop
                              ("# GET requests received (from other peers)"),
                            1,
                            GNUNET_NO);
  GSF_cover_query_count++;
  bm = ntohl (gm->hash_bitmap);
  bits = 0;
  if (0 != (bm & GET_MESSAGE_BIT_RETURN_TO))
    cp = GSF_peer_get_ (&opt[bits++]);
  else
    cp = cps;
  if (NULL == cp)
  {
    if (0 != (bm & GET_MESSAGE_BIT_RETURN_TO))
      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                  "Failed to find RETURN-TO peer `%s' in connection set. Dropping query.\n",
                  GNUNET_i2s (&opt[bits - 1]));
 
    else
      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                  "Failed to find peer `%s' in connection set. Dropping query.\n",
                  GNUNET_i2s (cps->ppd.peer));
    GNUNET_STATISTICS_update (GSF_stats,
                              gettext_noop
                              (
                                "# requests dropped due to missing reverse route"),
                              1,
                              GNUNET_NO);
    return;
  }
  unsigned int queue_size = GNUNET_MQ_get_length (cp->mq);
  queue_size += cp->ppd.pending_replies + cp->delay_queue_size;
  if (queue_size > MAX_QUEUE_PER_PEER)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Peer `%s' has too many replies queued already. Dropping query.\n",
                GNUNET_i2s (cps->ppd.peer));
    GNUNET_STATISTICS_update (GSF_stats,
                              gettext_noop (
                                "# requests dropped due to full reply queue"),
                              1,
                              GNUNET_NO);
    return;
  }
  /* note that we can really only check load here since otherwise
   * peers could find out that we are overloaded by not being
   * disconnected after sending us a malformed query... */
  tec.priority = bound_priority (ntohl (gm->priority),
                                 cps);
  if (tec.priority < 0)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Dropping query from `%s', this peer is too busy.\n",
                GNUNET_i2s (cps->ppd.peer));
    return;
  }
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Received request for `%s' of type %u from peer `%s' with flags %u\n",
              GNUNET_h2s (&gm->query),
              (unsigned int) tec.type,
              GNUNET_i2s (cps->ppd.peer),
              (unsigned int) bm);
  target =
    (0 !=
     (bm & GET_MESSAGE_BIT_TRANSMIT_TO)) ? (&opt[bits++]) : NULL;
  options = GSF_PRO_DEFAULTS;
  spid = 0;
  if ((GNUNET_LOAD_get_load (cp->ppd.transmission_delay) > 3 * (1
                                                                + tec.priority))
      || (GNUNET_LOAD_get_average (cp->ppd.transmission_delay) >
          GNUNET_CONSTANTS_MAX_CORK_DELAY.rel_value_us * 2
          + GNUNET_LOAD_get_average (GSF_rt_entry_lifetime)))
  {
    /* don't have BW to send to peer, or would likely take longer than we have for it,
     * so at best indirect the query */
    tec.priority = 0;
    options |= GSF_PRO_FORWARD_ONLY;
    spid = GNUNET_PEER_intern (cps->ppd.peer);
    GNUNET_assert (0 != spid);
  }
  tec.ttl = bound_ttl (ntohl (gm->ttl),
                       tec.priority);
  /* decrement ttl (always) */
  ttl_decrement =
    2 * TTL_DECREMENT + GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
                                                  TTL_DECREMENT);
  if ((tec.ttl < 0) &&
      (((int32_t) (tec.ttl - ttl_decrement)) > 0))
  {
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Dropping query from `%s' due to TTL underflow (%d - %u).\n",
                GNUNET_i2s (cps->ppd.peer),
                tec.ttl,
                ttl_decrement);
    GNUNET_STATISTICS_update (GSF_stats,
                              gettext_noop
                                ("# requests dropped due TTL underflow"), 1,
                              GNUNET_NO);
    /* integer underflow => drop (should be very rare)! */
    return;
  }
  tec.ttl -= ttl_decrement;
 
  /* test if the request already exists */
  tec.finished = GNUNET_NO;
  GNUNET_CONTAINER_multihashmap_get_multiple (cp->request_map,
                                              &gm->query,
                                              &test_exist_cb,
                                              &tec);
  if (GNUNET_YES == tec.finished)
    return; /* merged into existing request, we're done */
 
  peerreq = GNUNET_new (struct PeerRequest);
  peerreq->cp = cp;
  pr = GSF_pending_request_create_ (options,
                                    tec.type,
                                    &gm->query,
                                    target,
                                    (bfsize > 0)
                                    ? (const char *) &opt[bits]
                                    : NULL,
                                    bfsize,
                                    1 /* anonymity */,
                                    (uint32_t) tec.priority,
                                    tec.ttl,
                                    spid,
                                    GNUNET_PEER_intern (cps->ppd.peer),
                                    NULL, 0,        /* replies_seen */
                                    &handle_p2p_reply,
                                    peerreq);
  GNUNET_assert (NULL != pr);
  prd = GSF_pending_request_get_data_ (pr);
  peerreq->pr = pr;
  GNUNET_break (GNUNET_OK ==
                GNUNET_CONTAINER_multihashmap_put (cp->request_map,
                                                   &prd->query,
                                                   peerreq,
                                                   GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE));
  GNUNET_STATISTICS_update (GSF_stats,
                            gettext_noop (
                              "# P2P query messages received and processed"),
                            1,
                            GNUNET_NO);
  GNUNET_STATISTICS_update (GSF_stats,
                            gettext_noop ("# P2P searches active"),
                            1,
                            GNUNET_NO);
  GSF_pending_request_get_data_ (pr)->has_started = GNUNET_YES;
  GSF_local_lookup_ (pr,
                     &GSF_consider_forwarding,
                     NULL);
}
 
 
void
GSF_peer_transmit_ (struct GSF_ConnectedPeer *cp,
                    int is_query,
                    uint32_t priority,
                    struct GNUNET_MQ_Envelope *env)
{
  struct GSF_PeerTransmitHandle *pth;
  struct GSF_PeerTransmitHandle *pos;
  struct GSF_PeerTransmitHandle *prev;
 
  pth = GNUNET_new (struct GSF_PeerTransmitHandle);
  pth->transmission_request_start_time = GNUNET_TIME_absolute_get ();
  pth->env = env;
  pth->is_query = is_query;
  pth->priority = priority;
  pth->cp = cp;
  /* insertion sort (by priority, descending) */
  prev = NULL;
  pos = cp->pth_head;
  while ((NULL != pos) && (pos->priority > priority))
  {
    prev = pos;
    pos = pos->next;
  }
  GNUNET_CONTAINER_DLL_insert_after (cp->pth_head,
                                     cp->pth_tail,
                                     prev,
                                     pth);
  if (GNUNET_YES == is_query)
    cp->ppd.pending_queries++;
  else if (GNUNET_NO == is_query)
    cp->ppd.pending_replies++;
  schedule_transmission (pth);
}
 
 
void
GSF_peer_update_performance_ (struct GSF_ConnectedPeer *cp,
                              struct GNUNET_TIME_Absolute request_time,
                              uint32_t request_priority)
{
  struct GNUNET_TIME_Relative delay;
 
  delay = GNUNET_TIME_absolute_get_duration (request_time);
  cp->ppd.avg_reply_delay.rel_value_us =
    (cp->ppd.avg_reply_delay.rel_value_us * (RUNAVG_DELAY_N - 1)
     + delay.rel_value_us) / RUNAVG_DELAY_N;
  cp->ppd.avg_priority =
    (cp->ppd.avg_priority * (RUNAVG_DELAY_N - 1)
     + request_priority) / RUNAVG_DELAY_N;
}
 
 
void
GSF_peer_update_responder_client_ (struct GSF_ConnectedPeer *cp,
                                   struct GSF_LocalClient *initiator_client)
{
  cp->ppd.last_client_replies[cp->last_client_replies_woff++
                              % CS2P_SUCCESS_LIST_SIZE] = initiator_client;
}
 
 
void
GSF_peer_update_responder_peer_ (struct GSF_ConnectedPeer *cp,
                                 const struct GSF_ConnectedPeer *initiator_peer)
{
  unsigned int woff;
 
  woff = cp->last_p2p_replies_woff % P2P_SUCCESS_LIST_SIZE;
  GNUNET_PEER_change_rc (cp->ppd.last_p2p_replies[woff], -1);
  cp->ppd.last_p2p_replies[woff] = initiator_peer->ppd.pid;
  GNUNET_PEER_change_rc (initiator_peer->ppd.pid, 1);
  cp->last_p2p_replies_woff = (woff + 1) % P2P_SUCCESS_LIST_SIZE;
}
 
 
static int
flush_respect (void *cls,
               const struct GNUNET_PeerIdentity *key,
               void *value)
{
  struct GSF_ConnectedPeer *cp = value;
  struct GNUNET_PeerIdentity pid;
 
  if (cp->ppd.respect == cp->disk_respect)
    return GNUNET_OK;           /* unchanged */
  GNUNET_assert (0 != cp->ppd.pid);
  GNUNET_PEER_resolve (cp->ppd.pid, &pid);
  GNUNET_PEERSTORE_store (peerstore, "fs", &pid, "respect", &cp->ppd.respect,
                          sizeof(cp->ppd.respect),
                          GNUNET_TIME_UNIT_FOREVER_ABS,
                          GNUNET_PEERSTORE_STOREOPTION_REPLACE,
                          NULL,
                          NULL);
  return GNUNET_OK;
}
 
 
void
GSF_peer_disconnect_handler (void *cls,
                             const struct GNUNET_PeerIdentity *peer,
                             void *internal_cls)
{
  struct GSF_ConnectedPeer *cp = internal_cls;
  struct GSF_PeerTransmitHandle *pth;
  struct GSF_DelayedHandle *dh;
 
  if (NULL == cp)
    return;  /* must have been disconnect from core with
              * 'peer' == my_id, ignore */
  flush_respect (NULL,
                 peer,
                 cp);
  GNUNET_assert (GNUNET_YES ==
                 GNUNET_CONTAINER_multipeermap_remove (cp_map,
                                                       peer,
                                                       cp));
  GNUNET_STATISTICS_set (GSF_stats,
                         gettext_noop ("# peers connected"),
                         GNUNET_CONTAINER_multipeermap_size (cp_map),
                         GNUNET_NO);
  if (NULL != cp->respect_iterate_req)
  {
    GNUNET_PEERSTORE_iteration_stop (cp->respect_iterate_req);
    cp->respect_iterate_req = NULL;
  }
  GNUNET_CONTAINER_multihashmap_iterate (cp->request_map,
                                         &cancel_pending_request,
                                         cp);
  GNUNET_CONTAINER_multihashmap_destroy (cp->request_map);
  cp->request_map = NULL;
  GSF_plan_notify_peer_disconnect_ (cp);
  GNUNET_LOAD_value_free (cp->ppd.transmission_delay);
  GNUNET_PEER_decrement_rcs (cp->ppd.last_p2p_replies,
                             P2P_SUCCESS_LIST_SIZE);
  memset (cp->ppd.last_p2p_replies,
          0,
          sizeof(cp->ppd.last_p2p_replies));
  GSF_push_stop_ (cp);
  while (NULL != (pth = cp->pth_head))
  {
    GNUNET_CONTAINER_DLL_remove (cp->pth_head,
                                 cp->pth_tail,
                                 pth);
    if (GNUNET_YES == pth->is_query)
      GNUNET_assert (0 < cp->ppd.pending_queries--);
    else if (GNUNET_NO == pth->is_query)
      GNUNET_assert (0 < cp->ppd.pending_replies--);
    GNUNET_free (pth);
  }
  while (NULL != (dh = cp->delayed_head))
  {
    GNUNET_CONTAINER_DLL_remove (cp->delayed_head,
                                 cp->delayed_tail,
                                 dh);
    GNUNET_MQ_discard (dh->env);
    cp->delay_queue_size--;
    GNUNET_SCHEDULER_cancel (dh->delay_task);
    GNUNET_free (dh);
  }
  GNUNET_PEER_change_rc (cp->ppd.pid, -1);
  if (NULL != cp->mig_revive_task)
  {
    GNUNET_SCHEDULER_cancel (cp->mig_revive_task);
    cp->mig_revive_task = NULL;
  }
  GNUNET_break (0 == cp->ppd.pending_queries);
  GNUNET_break (0 == cp->ppd.pending_replies);
  GNUNET_free (cp);
}
 
 
struct IterationContext
{
  GSF_ConnectedPeerIterator it;
 
  void *it_cls;
};
 
 
static int
call_iterator (void *cls,
               const struct GNUNET_PeerIdentity *key,
               void *value)
{
  struct IterationContext *ic = cls;
  struct GSF_ConnectedPeer *cp = value;
 
  ic->it (ic->it_cls,
          key, cp,
          &cp->ppd);
  return GNUNET_YES;
}
 
 
void
GSF_iterate_connected_peers_ (GSF_ConnectedPeerIterator it,
                              void *it_cls)
{
  struct IterationContext ic;
 
  ic.it = it;
  ic.it_cls = it_cls;
  GNUNET_CONTAINER_multipeermap_iterate (cp_map,
                                         &call_iterator,
                                         &ic);
}
 
 
void
GSF_connected_peer_get_identity_ (const struct GSF_ConnectedPeer *cp,
                                  struct GNUNET_PeerIdentity *id)
{
  GNUNET_assert (0 != cp->ppd.pid);
  GNUNET_PEER_resolve (cp->ppd.pid, id);
}
 
 
const struct GNUNET_PeerIdentity *
GSF_connected_peer_get_identity2_ (const struct GSF_ConnectedPeer *cp)
{
  GNUNET_assert (0 != cp->ppd.pid);
  return GNUNET_PEER_resolve2 (cp->ppd.pid);
}
 
 
void
GSF_block_peer_migration_ (struct GSF_ConnectedPeer *cp,
                           struct GNUNET_TIME_Absolute block_time)
{
  struct GNUNET_MQ_Envelope *env;
  struct MigrationStopMessage *msm;
 
  if (cp->last_migration_block.abs_value_us > block_time.abs_value_us)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Migration already blocked for another %s\n",
                GNUNET_STRINGS_relative_time_to_string (
                  GNUNET_TIME_absolute_get_remaining
                    (cp->
                    last_migration_block), GNUNET_YES));
    return;                     /* already blocked */
  }
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Asking to stop migration for %s\n",
              GNUNET_STRINGS_relative_time_to_string (
                GNUNET_TIME_absolute_get_remaining (block_time),
                GNUNET_YES));
  cp->last_migration_block = block_time;
  env = GNUNET_MQ_msg (msm,
                       GNUNET_MESSAGE_TYPE_FS_MIGRATION_STOP);
  msm->reserved = htonl (0);
  msm->duration
    = GNUNET_TIME_relative_hton (GNUNET_TIME_absolute_get_remaining
                                   (cp->last_migration_block));
  GNUNET_STATISTICS_update (GSF_stats,
                            gettext_noop ("# migration stop messages sent"),
                            1,
                            GNUNET_NO);
  GSF_peer_transmit_ (cp,
                      GNUNET_SYSERR,
                      UINT32_MAX,
                      env);
}
 
 
void
GSF_connected_peer_change_preference_ (struct GSF_ConnectedPeer *cp,
                                       uint64_t pref)
{
  cp->inc_preference += pref;
}
 
 
static void
cron_flush_respect (void *cls)
{
  fr_task = NULL;
  GNUNET_CONTAINER_multipeermap_iterate (cp_map,
                                         &flush_respect,
                                         NULL);
  fr_task = GNUNET_SCHEDULER_add_delayed_with_priority (RESPECT_FLUSH_FREQ,
                                                        GNUNET_SCHEDULER_PRIORITY_HIGH,
                                                        &cron_flush_respect,
                                                        NULL);
}
 
 
void
GSF_connected_peer_init_ ()
{
  cp_map = GNUNET_CONTAINER_multipeermap_create (128, GNUNET_YES);
  peerstore = GNUNET_PEERSTORE_connect (GSF_cfg);
  fr_task = GNUNET_SCHEDULER_add_with_priority (GNUNET_SCHEDULER_PRIORITY_HIGH,
                                                &cron_flush_respect, NULL);
}
 
 
void
GSF_connected_peer_done_ ()
{
  GNUNET_CONTAINER_multipeermap_iterate (cp_map,
                                         &flush_respect,
                                         NULL);
  GNUNET_SCHEDULER_cancel (fr_task);
  fr_task = NULL;
  GNUNET_CONTAINER_multipeermap_destroy (cp_map);
  cp_map = NULL;
  GNUNET_PEERSTORE_disconnect (peerstore);
}
 
 
static int
clean_local_client (void *cls,
                    const struct GNUNET_PeerIdentity *key,
                    void *value)
{
  const struct GSF_LocalClient *lc = cls;
  struct GSF_ConnectedPeer *cp = value;
  unsigned int i;
 
  for (i = 0; i < CS2P_SUCCESS_LIST_SIZE; i++)
    if (cp->ppd.last_client_replies[i] == lc)
      cp->ppd.last_client_replies[i] = NULL;
  return GNUNET_YES;
}
 
 
void
GSF_handle_local_client_disconnect_ (const struct GSF_LocalClient *lc)
{
  if (NULL == cp_map)
    return;                     /* already cleaned up */
  GNUNET_CONTAINER_multipeermap_iterate (cp_map,
                                         &clean_local_client,
                                         (void *) lc);
}
 
 
/* end of gnunet-service-fs_cp.c */

REPLY_TIMEOUT

#define REPLY_TIMEOUT

Value:

                                                     GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MINUTES, \
                                                     2)

After how long do we discard a reply?

Definition at line 54 of file gnunet-service-fs_cp.c.

INSANE_STATISTICS

#define INSANE_STATISTICS   GNUNET_NO

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

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

N

#define N   ((double) 128.0)

Function Documentation

GSF_update_peer_latency_()

void GSF_update_peer_latency_	(	const struct GNUNET_PeerIdentity *	id,
		struct GNUNET_TIME_Relative	latency
	)

Update the latency information kept for the given peer.

Parameters

id	peer record to update
latency	current latency value

Definition at line 288 of file gnunet-service-fs_cp.c.

{
  struct GSF_ConnectedPeer *cp;
 
  cp = GSF_peer_get_ (id);
  if (NULL == cp)
    return; /* we're not yet connected at the core level, ignore */
  GNUNET_LOAD_value_set_decline (cp->ppd.transmission_delay,
                                 latency);
}

References GNUNET_LOAD_value_set_decline(), GSF_peer_get_(), GSF_ConnectedPeer::ppd, and GSF_PeerPerformanceData::transmission_delay.

Here is the call graph for this function:

GSF_get_peer_performance_data_()

struct GSF_PeerPerformanceData * GSF_get_peer_performance_data_

(

struct GSF_ConnectedPeer *

cp

)

Return the performance data record for the given peer.

Parameters

cp	peer to query

Returns

performance data record for the peer

Definition at line 308 of file gnunet-service-fs_cp.c.

{
  return &cp->ppd;
}

References GSF_ConnectedPeer::ppd.

Referenced by handle_p2p_put(), put_migration_continuation(), score_content(), and transmit_content().

Here is the caller graph for this function:

peer_transmit()

static void peer_transmit ( struct GSF_ConnectedPeer *  cp )

static

Core is ready to transmit to a peer, get the message.

Parameters

cp	which peer to send a message to

Definition at line 349 of file gnunet-service-fs_cp.c.

{
  struct GSF_PeerTransmitHandle *pth = cp->pth_head;
  struct GSF_PeerTransmitHandle *pos;
 
  if (NULL == pth)
    return;
  GNUNET_CONTAINER_DLL_remove (cp->pth_head,
                               cp->pth_tail,
                               pth);
  if (GNUNET_YES == pth->is_query)
  {
    cp->ppd.last_request_times[(cp->last_request_times_off++)
                               % MAX_QUEUE_PER_PEER] =
      GNUNET_TIME_absolute_get ();
    GNUNET_assert (0 < cp->ppd.pending_queries--);
  }
  else if (GNUNET_NO == pth->is_query)
  {
    GNUNET_assert (0 < cp->ppd.pending_replies--);
  }
  GNUNET_LOAD_update (cp->ppd.transmission_delay,
                      GNUNET_TIME_absolute_get_duration
                        (pth->transmission_request_start_time).rel_value_us);
  GNUNET_MQ_send (cp->mq,
                  pth->env);
  GNUNET_free (pth);
  if (NULL != (pos = cp->pth_head))
  {
    GNUNET_assert (pos != pth);
    schedule_transmission (pos);
  }
}

References GSF_PeerTransmitHandle::cp, GSF_PeerTransmitHandle::env, GNUNET_assert, GNUNET_CONTAINER_DLL_remove, GNUNET_free, GNUNET_LOAD_update(), GNUNET_MQ_send(), GNUNET_NO, GNUNET_TIME_absolute_get(), GNUNET_TIME_absolute_get_duration(), GNUNET_YES, GSF_PeerTransmitHandle::is_query, GSF_PeerPerformanceData::last_request_times, GSF_ConnectedPeer::last_request_times_off, MAX_QUEUE_PER_PEER, GSF_ConnectedPeer::mq, GSF_PeerPerformanceData::pending_queries, GSF_PeerPerformanceData::pending_replies, GSF_ConnectedPeer::ppd, GSF_ConnectedPeer::pth_head, GSF_ConnectedPeer::pth_tail, GNUNET_TIME_Relative::rel_value_us, schedule_transmission(), GSF_PeerPerformanceData::transmission_delay, and GSF_PeerTransmitHandle::transmission_request_start_time.

Referenced by schedule_transmission().

Here is the call graph for this function:

Here is the caller graph for this function:

schedule_transmission()

static void schedule_transmission ( struct GSF_PeerTransmitHandle *  pth )

static

If ready (bandwidth reserved), try to schedule transmission via core for the given handle.

Parameters

pth	transmission handle to schedule

Definition at line 330 of file gnunet-service-fs_cp.c.

{
  struct GSF_ConnectedPeer *cp;
  struct GNUNET_PeerIdentity target;
 
  cp = pth->cp;
  GNUNET_assert (0 != cp->ppd.pid);
  GNUNET_PEER_resolve (cp->ppd.pid, &target);
 
  peer_transmit (cp);
}

References GSF_PeerTransmitHandle::cp, GNUNET_assert, GNUNET_PEER_resolve(), peer_transmit(), GSF_PeerPerformanceData::pid, and GSF_ConnectedPeer::ppd.

Referenced by GSF_peer_transmit_(), and peer_transmit().

Here is the call graph for this function:

Here is the caller graph for this function:

peer_respect_cb()

static void peer_respect_cb ( void *  cls, const struct GNUNET_PEERSTORE_Record *  record, const char *  emsg  )

static

Function called by PEERSTORE with peer respect record.

Parameters

cls	handle to connected peer entry
record	peerstore record information
emsg	error message, or NULL if no errors

Definition at line 392 of file gnunet-service-fs_cp.c.

{
  struct GSF_ConnectedPeer *cp = cls;
 
  GNUNET_assert (NULL != cp->respect_iterate_req);
  if ((NULL == record) && (NULL == emsg))
  {
    cp->respect_iterate_req = NULL;
    return;
  }
  if ((NULL != record) &&
      (sizeof(cp->disk_respect) == record->value_size))
  {
    cp->disk_respect = *((uint32_t *) record->value);
    cp->ppd.respect += *((uint32_t *) record->value);
  }
  GSF_push_start_ (cp);
  if (NULL != record)
  {
    GNUNET_PEERSTORE_iteration_stop (cp->respect_iterate_req);
    cp->respect_iterate_req = NULL;
    return;
  }
  GNUNET_PEERSTORE_iteration_next (cp->respect_iterate_req, 1);
}

References GSF_ConnectedPeer::disk_respect, GNUNET_assert, GNUNET_PEERSTORE_iteration_next(), GNUNET_PEERSTORE_iteration_stop(), GSF_push_start_(), GSF_ConnectedPeer::ppd, record(), GSF_PeerPerformanceData::respect, and GSF_ConnectedPeer::respect_iterate_req.

Referenced by GSF_peer_connect_handler().

Here is the call graph for this function:

Here is the caller graph for this function:

consider_peer_for_forwarding()

static int consider_peer_for_forwarding ( void *  cls, const struct GNUNET_HashCode *  key, struct GSF_PendingRequest *  pr  )

static

Function called for each pending request whenever a new peer connects, giving us a chance to decide about submitting the existing request to the new peer.

Parameters

cls	the struct GSF_ConnectedPeer of the new peer
key	query for the request
pr	handle to the pending request

Returns

GNUNET_YES to continue to iterate

Definition at line 432 of file gnunet-service-fs_cp.c.

{
  struct GSF_ConnectedPeer *cp = cls;
  struct GNUNET_PeerIdentity pid;
 
  if (GNUNET_YES !=
      GSF_pending_request_test_active_ (pr))
    return GNUNET_YES; /* request is not actually active, skip! */
  GSF_connected_peer_get_identity_ (cp, &pid);
  if (GNUNET_YES !=
      GSF_pending_request_test_target_ (pr, &pid))
  {
    GNUNET_STATISTICS_update (GSF_stats,
                              gettext_noop ("# Loopback routes suppressed"),
                              1,
                              GNUNET_NO);
    return GNUNET_YES;
  }
  GSF_plan_add_ (cp, pr);
  return GNUNET_YES;
}

References gettext_noop, GNUNET_NO, GNUNET_STATISTICS_update(), GNUNET_YES, GSF_connected_peer_get_identity_(), GSF_pending_request_test_active_(), GSF_pending_request_test_target_(), GSF_plan_add_(), GSF_stats, and pid.

Referenced by GSF_peer_connect_handler().

Here is the call graph for this function:

Here is the caller graph for this function:

GSF_peer_connect_handler()

void * GSF_peer_connect_handler	(	void *	cls,
		const struct GNUNET_PeerIdentity *	peer,
		struct GNUNET_MQ_Handle *	mq,
		enum GNUNET_CORE_PeerClass	class
	)

A peer connected to us.

Setup the connected peer records.

Parameters

cls	NULL
peer	identity of peer that connected
mq	message queue for talking to peer
class	peer class of connected peer

Returns

internal handle for the peer

Definition at line 458 of file gnunet-service-fs_cp.c.

{
  const struct GNUNET_PeerIdentity *my_identity;
  struct GSF_ConnectedPeer *cp;
 
  my_identity = GNUNET_PILS_get_identity (GSF_pils);
  GNUNET_assert (my_identity);
 
  if (0 == GNUNET_memcmp (my_identity, peer))
    return NULL;
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Connected to peer %s\n",
              GNUNET_i2s (peer));
  cp = GNUNET_new (struct GSF_ConnectedPeer);
  cp->ppd.pid = GNUNET_PEER_intern (peer);
  cp->ppd.peer = peer;
  cp->mq = mq;
  cp->ppd.transmission_delay = GNUNET_LOAD_value_init (GNUNET_TIME_UNIT_ZERO);
 
  cp->request_map = GNUNET_CONTAINER_multihashmap_create (128,
                                                          GNUNET_YES);
  GNUNET_break (GNUNET_OK ==
                GNUNET_CONTAINER_multipeermap_put (cp_map,
                                                   GSF_connected_peer_get_identity2_
                                                   (
                                                     cp),
                                                   cp,
                                                   GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
  GNUNET_STATISTICS_set (GSF_stats,
                         gettext_noop ("# peers connected"),
                         GNUNET_CONTAINER_multipeermap_size (cp_map),
                         GNUNET_NO);
  cp->respect_iterate_req
    = GNUNET_PEERSTORE_iteration_start (peerstore,
                                        "fs",
                                        peer,
                                        "respect",
                                        &peer_respect_cb,
                                        cp);
  GSF_iterate_pending_requests_ (&consider_peer_for_forwarding,
                                 cp);
  return cp;
}

References consider_peer_for_forwarding(), cp_map, gettext_noop, GNUNET_assert, GNUNET_break, GNUNET_CONTAINER_multihashmap_create(), GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY, GNUNET_CONTAINER_multipeermap_put(), GNUNET_CONTAINER_multipeermap_size(), GNUNET_ERROR_TYPE_DEBUG, GNUNET_i2s(), GNUNET_LOAD_value_init(), GNUNET_log, GNUNET_memcmp, GNUNET_new, GNUNET_NO, GNUNET_OK, GNUNET_PEER_intern(), GNUNET_PEERSTORE_iteration_start(), GNUNET_PILS_get_identity(), GNUNET_STATISTICS_set(), GNUNET_TIME_UNIT_ZERO, GNUNET_YES, GSF_connected_peer_get_identity2_(), GSF_iterate_pending_requests_(), GSF_pils, GSF_stats, mq, GSF_ConnectedPeer::mq, my_identity, GSF_PeerPerformanceData::peer, peer_respect_cb(), peerstore, GSF_PeerPerformanceData::pid, GSF_ConnectedPeer::ppd, GSF_ConnectedPeer::request_map, GSF_ConnectedPeer::respect_iterate_req, and GSF_PeerPerformanceData::transmission_delay.

Referenced by main_init().

Here is the call graph for this function:

Here is the caller graph for this function:

revive_migration()

static void revive_migration ( void *  cls )

static

It may be time to re-start migrating content to this peer.

Check, and if so, restart migration.

Parameters

cls	the struct GSF_ConnectedPeer

Definition at line 513 of file gnunet-service-fs_cp.c.

{
  struct GSF_ConnectedPeer *cp = cls;
  struct GNUNET_TIME_Relative bt;
 
  cp->mig_revive_task = NULL;
  bt = GNUNET_TIME_absolute_get_remaining (cp->ppd.migration_blocked_until);
  if (0 != bt.rel_value_us)
  {
    /* still time left... */
    cp->mig_revive_task =
      GNUNET_SCHEDULER_add_delayed (bt, &revive_migration, cp);
    return;
  }
  GSF_push_start_ (cp);
}

References GNUNET_SCHEDULER_add_delayed(), GNUNET_TIME_absolute_get_remaining(), GSF_push_start_(), GSF_ConnectedPeer::mig_revive_task, GSF_PeerPerformanceData::migration_blocked_until, GSF_ConnectedPeer::ppd, GNUNET_TIME_Relative::rel_value_us, and revive_migration().

Referenced by handle_p2p_migration_stop(), and revive_migration().

Here is the call graph for this function:

Here is the caller graph for this function:

GSF_peer_get_()

struct GSF_ConnectedPeer * GSF_peer_get_

(

const struct GNUNET_PeerIdentity *

peer

)

Get a handle for a connected peer.

Parameters

peer	peer's identity

Returns

NULL if this peer is not currently connected

Definition at line 532 of file gnunet-service-fs_cp.c.

{
  if (NULL == cp_map)
    return NULL;
  return GNUNET_CONTAINER_multipeermap_get (cp_map, peer);
}

References cp_map, and GNUNET_CONTAINER_multipeermap_get().

Referenced by GSF_update_peer_latency_(), handle_p2p_get(), and put_migration_continuation().

Here is the call graph for this function:

Here is the caller graph for this function:

handle_p2p_migration_stop()

void handle_p2p_migration_stop	(	void *	cls,
		const struct MigrationStopMessage *	msm
	)

Handle P2P GNUNET_MESSAGE_TYPE_FS_MIGRATION_STOP message.

Parameters

cls	closure, the struct GSF_ConnectedPeer
msm	the actual message

Definition at line 547 of file gnunet-service-fs_cp.c.

{
  struct GSF_ConnectedPeer *cp = cls;
  struct GNUNET_TIME_Relative bt;
 
  GNUNET_STATISTICS_update (GSF_stats,
                            gettext_noop ("# migration stop messages received"),
                            1, GNUNET_NO);
  bt = GNUNET_TIME_relative_ntoh (msm->duration);
  GNUNET_log (GNUNET_ERROR_TYPE_INFO,
              _ ("Migration of content to peer `%s' blocked for %s\n"),
              GNUNET_i2s (cp->ppd.peer),
              GNUNET_STRINGS_relative_time_to_string (bt, GNUNET_YES));
  cp->ppd.migration_blocked_until = GNUNET_TIME_relative_to_absolute (bt);
  if ((NULL == cp->mig_revive_task) &&
      (NULL == cp->respect_iterate_req))
  {
    GSF_push_stop_ (cp);
    cp->mig_revive_task =
      GNUNET_SCHEDULER_add_delayed (bt,
                                    &revive_migration, cp);
  }
}

References _, MigrationStopMessage::duration, gettext_noop, GNUNET_ERROR_TYPE_INFO, GNUNET_i2s(), GNUNET_log, GNUNET_NO, GNUNET_SCHEDULER_add_delayed(), GNUNET_STATISTICS_update(), GNUNET_STRINGS_relative_time_to_string(), GNUNET_TIME_relative_ntoh(), GNUNET_TIME_relative_to_absolute(), GNUNET_YES, GSF_push_stop_(), GSF_stats, GSF_ConnectedPeer::mig_revive_task, GSF_PeerPerformanceData::migration_blocked_until, GSF_PeerPerformanceData::peer, GSF_ConnectedPeer::ppd, GSF_ConnectedPeer::respect_iterate_req, and revive_migration().

Here is the call graph for this function:

free_pending_request()

static void free_pending_request ( struct PeerRequest *  peerreq )

static

Free resources associated with the given peer request.

Parameters

peerreq

request to free

Definition at line 579 of file gnunet-service-fs_cp.c.

{
  struct GSF_ConnectedPeer *cp = peerreq->cp;
  struct GSF_PendingRequestData *prd;
 
  prd = GSF_pending_request_get_data_ (peerreq->pr);
  if (NULL != peerreq->kill_task)
  {
    GNUNET_SCHEDULER_cancel (peerreq->kill_task);
    peerreq->kill_task = NULL;
  }
  GNUNET_STATISTICS_update (GSF_stats,
                            gettext_noop ("# P2P searches active"),
                            -1,
                            GNUNET_NO);
  GNUNET_break (GNUNET_YES ==
                GNUNET_CONTAINER_multihashmap_remove (cp->request_map,
                                                      &prd->query,
                                                      peerreq));
  GNUNET_free (peerreq);
}

References PeerRequest::cp, gettext_noop, GNUNET_break, GNUNET_CONTAINER_multihashmap_remove(), GNUNET_free, GNUNET_NO, GNUNET_SCHEDULER_cancel(), GNUNET_STATISTICS_update(), GNUNET_YES, GSF_pending_request_get_data_(), GSF_stats, PeerRequest::kill_task, PeerRequest::pr, GSF_PendingRequestData::query, and GSF_ConnectedPeer::request_map.

Referenced by cancel_pending_request(), handle_p2p_reply(), and test_exist_cb().

Here is the call graph for this function:

Here is the caller graph for this function:

cancel_pending_request()

static int cancel_pending_request ( void *  cls, const struct GNUNET_HashCode *  query, void *  value  )

static

Cancel all requests associated with the peer.

Parameters

cls	unused
query	hash code of the request
value	the struct GSF_PendingRequest

Returns

GNUNET_YES (continue to iterate)

Definition at line 611 of file gnunet-service-fs_cp.c.

{
  struct PeerRequest *peerreq = value;
  struct GSF_PendingRequest *pr = peerreq->pr;
 
  free_pending_request (peerreq);
  GSF_pending_request_cancel_ (pr,
                               GNUNET_NO);
  return GNUNET_OK;
}

References free_pending_request(), GNUNET_NO, GNUNET_OK, GSF_pending_request_cancel_(), PeerRequest::pr, and value.

Referenced by GSF_peer_disconnect_handler(), and peer_request_destroy().

Here is the call graph for this function:

Here is the caller graph for this function:

peer_request_destroy()

static void peer_request_destroy ( void *  cls )

static

Free the given request.

Parameters

cls	the request to free

Definition at line 631 of file gnunet-service-fs_cp.c.

{
  struct PeerRequest *peerreq = cls;
  struct GSF_PendingRequest *pr = peerreq->pr;
  struct GSF_PendingRequestData *prd;
 
  peerreq->kill_task = NULL;
  prd = GSF_pending_request_get_data_ (pr);
  cancel_pending_request (NULL,
                          &prd->query,
                          peerreq);
}

References cancel_pending_request(), GSF_pending_request_get_data_(), PeerRequest::kill_task, PeerRequest::pr, and GSF_PendingRequestData::query.

Referenced by handle_p2p_reply().

Here is the call graph for this function:

Here is the caller graph for this function:

transmit_delayed_now()

static void transmit_delayed_now ( void *  cls )

static

The artificial delay is over, transmit the message now.

Parameters

cls	the struct GSF_DelayedHandle with the message

Definition at line 651 of file gnunet-service-fs_cp.c.

{
  struct GSF_DelayedHandle *dh = cls;
  struct GSF_ConnectedPeer *cp = dh->cp;
 
  GNUNET_CONTAINER_DLL_remove (cp->delayed_head,
                               cp->delayed_tail,
                               dh);
  cp->delay_queue_size--;
  GSF_peer_transmit_ (cp,
                      GNUNET_NO,
                      UINT32_MAX,
                      dh->env);
  GNUNET_free (dh);
}

References GSF_DelayedHandle::cp, GSF_ConnectedPeer::delay_queue_size, GSF_ConnectedPeer::delayed_head, GSF_ConnectedPeer::delayed_tail, GSF_DelayedHandle::env, GNUNET_CONTAINER_DLL_remove, GNUNET_free, GNUNET_NO, and GSF_peer_transmit_().

Referenced by handle_p2p_reply().

Here is the call graph for this function:

Here is the caller graph for this function:

get_randomized_delay()

static struct GNUNET_TIME_Relative get_randomized_delay ( )

static

Get the randomized delay a response should be subjected to.

Returns

desired delay

Definition at line 674 of file gnunet-service-fs_cp.c.

{
  struct GNUNET_TIME_Relative ret;
 
  ret =
    GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MILLISECONDS,
                                   GNUNET_CRYPTO_random_u32
                                     (GNUNET_CRYPTO_QUALITY_WEAK,
                                     2 * GSF_avg_latency.rel_value_us + 1));
#if INSANE_STATISTICS
  GNUNET_STATISTICS_update (GSF_stats,
                            gettext_noop
                              ("# artificial delays introduced (ms)"),
                            ret.rel_value_us / 1000LL, GNUNET_NO);
#endif
  return ret;
}

References gettext_noop, GNUNET_CRYPTO_QUALITY_WEAK, GNUNET_CRYPTO_random_u32(), GNUNET_NO, GNUNET_STATISTICS_update(), GNUNET_TIME_relative_multiply(), GNUNET_TIME_UNIT_MILLISECONDS, GSF_avg_latency, GSF_stats, GNUNET_TIME_Relative::rel_value_us, and ret.

Referenced by handle_p2p_reply().

Here is the call graph for this function:

Here is the caller graph for this function:

handle_p2p_reply()

static void handle_p2p_reply ( void *  cls, enum GNUNET_BLOCK_ReplyEvaluationResult  eval, struct GSF_PendingRequest *  pr, uint32_t  reply_anonymity_level, struct GNUNET_TIME_Absolute  expiration, struct GNUNET_TIME_Absolute  last_transmission, enum GNUNET_BLOCK_Type  type, const void *  data, size_t  data_len  )

static

Handle a reply to a pending request.

Also called if a request expires (then with data == NULL). The handler may be called many times (depending on the request type), but will not be called during or after a call to GSF_pending_request_cancel and will also not be called anymore after a call signalling expiration.

Parameters

cls	struct PeerRequest this is an answer for
eval	evaluation of the result
pr	handle to the original pending request
reply_anonymity_level	anonymity level for the reply, UINT32_MAX for "unknown"
expiration	when does data expire?
last_transmission	when did we last transmit a request for this block
type	type of the block
data	response data, NULL on request expiration
data_len	number of bytes in data

Definition at line 712 of file gnunet-service-fs_cp.c.

{
  struct PeerRequest *peerreq = cls;
  struct GSF_ConnectedPeer *cp = peerreq->cp;
  struct GSF_PendingRequestData *prd;
  struct GNUNET_MQ_Envelope *env;
  struct PutMessage *pm;
  size_t msize;
 
  GNUNET_assert (data_len + sizeof(struct PutMessage) <
                 GNUNET_MAX_MESSAGE_SIZE);
  GNUNET_assert (peerreq->pr == pr);
  prd = GSF_pending_request_get_data_ (pr);
  if (NULL == data)
  {
    free_pending_request (peerreq);
    return;
  }
  GNUNET_break (GNUNET_BLOCK_TYPE_ANY != type);
  if ( (prd->type != type) &&
       (GNUNET_BLOCK_TYPE_ANY != prd->type) )
  {
    GNUNET_STATISTICS_update (GSF_stats,
                              "# replies dropped due to type mismatch",
                              1, GNUNET_NO);
    return;
  }
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Transmitting result for query `%s' to peer\n",
              GNUNET_h2s (&prd->query));
  GNUNET_STATISTICS_update (GSF_stats,
                            "# replies received for other peers",
                            1,
                            GNUNET_NO);
  msize = sizeof(struct PutMessage) + data_len;
  if (msize >= GNUNET_MAX_MESSAGE_SIZE)
  {
    GNUNET_break (0);
    return;
  }
  if ( (UINT32_MAX != reply_anonymity_level) &&
       (reply_anonymity_level > 1) )
  {
    if (reply_anonymity_level - 1 > GSF_cover_content_count)
    {
      GNUNET_STATISTICS_update (GSF_stats,
                                "# replies dropped due to insufficient cover traffic",
                                1, GNUNET_NO);
      return;
    }
    GSF_cover_content_count -= (reply_anonymity_level - 1);
  }
 
  env = GNUNET_MQ_msg_extra (pm,
                             data_len,
                             GNUNET_MESSAGE_TYPE_FS_PUT);
  pm->type = htonl (type);
  pm->expiration = GNUNET_TIME_absolute_hton (expiration);
  GNUNET_memcpy (&pm[1],
                 data,
                 data_len);
  if ((UINT32_MAX != reply_anonymity_level) &&
      (0 != reply_anonymity_level) &&
      (GNUNET_YES == GSF_enable_randomized_delays))
  {
    struct GSF_DelayedHandle *dh;
 
    dh = GNUNET_new (struct GSF_DelayedHandle);
    dh->cp = cp;
    dh->env = env;
    dh->msize = msize;
    GNUNET_CONTAINER_DLL_insert (cp->delayed_head,
                                 cp->delayed_tail,
                                 dh);
    cp->delay_queue_size++;
    dh->delay_task =
      GNUNET_SCHEDULER_add_delayed (get_randomized_delay (),
                                    &transmit_delayed_now,
                                    dh);
  }
  else
  {
    GSF_peer_transmit_ (cp,
                        GNUNET_NO,
                        UINT32_MAX,
                        env);
  }
  if (GNUNET_BLOCK_REPLY_OK_LAST != eval)
    return;
  if (NULL == peerreq->kill_task)
  {
    GNUNET_STATISTICS_update (GSF_stats,
                              "# P2P searches destroyed due to ultimate reply",
                              1,
                              GNUNET_NO);
    peerreq->kill_task =
      GNUNET_SCHEDULER_add_now (&peer_request_destroy,
                                peerreq);
  }
}

References GSF_DelayedHandle::cp, PeerRequest::cp, data, GSF_ConnectedPeer::delay_queue_size, GSF_DelayedHandle::delay_task, GSF_ConnectedPeer::delayed_head, GSF_ConnectedPeer::delayed_tail, env, GSF_DelayedHandle::env, expiration, free_pending_request(), get_randomized_delay(), GNUNET_assert, GNUNET_BLOCK_REPLY_OK_LAST, GNUNET_BLOCK_TYPE_ANY, GNUNET_break, GNUNET_CONTAINER_DLL_insert, GNUNET_ERROR_TYPE_DEBUG, GNUNET_h2s(), GNUNET_log, GNUNET_MAX_MESSAGE_SIZE, GNUNET_memcpy, GNUNET_MESSAGE_TYPE_FS_PUT, GNUNET_MQ_msg_extra, GNUNET_new, GNUNET_NO, GNUNET_SCHEDULER_add_delayed(), GNUNET_SCHEDULER_add_now(), GNUNET_STATISTICS_update(), GNUNET_TIME_absolute_hton(), GNUNET_YES, GSF_cover_content_count, GSF_enable_randomized_delays, GSF_peer_transmit_(), GSF_pending_request_get_data_(), GSF_stats, PeerRequest::kill_task, GSF_DelayedHandle::msize, peer_request_destroy(), pm, PeerRequest::pr, GSF_PendingRequestData::query, transmit_delayed_now(), type, and GSF_PendingRequestData::type.

Referenced by handle_p2p_get().

Here is the call graph for this function:

Here is the caller graph for this function:

change_peer_respect()

static int change_peer_respect ( struct GSF_ConnectedPeer *  cp, int  value  )

static

Increase the peer's respect by a value.

Parameters

cp	which peer to change the respect value on
value	is the int value by which the peer's credit is to be increased or decreased

Returns

the actual change in respect (positive or negative)

Definition at line 831 of file gnunet-service-fs_cp.c.

{
  if (0 == value)
    return 0;
  GNUNET_assert (NULL != cp);
  if (value > 0)
  {
    if (cp->ppd.respect + value < cp->ppd.respect)
    {
      value = UINT32_MAX - cp->ppd.respect;
      cp->ppd.respect = UINT32_MAX;
    }
    else
      cp->ppd.respect += value;
  }
  else
  {
    if (cp->ppd.respect < -value)
    {
      value = -cp->ppd.respect;
      cp->ppd.respect = 0;
    }
    else
      cp->ppd.respect += value;
  }
  return value;
}

References GSF_DelayedHandle::cp, GNUNET_assert, GSF_ConnectedPeer::ppd, GSF_PeerPerformanceData::respect, and value.

Referenced by bound_priority().

Here is the caller graph for this function:

bound_priority()

static int32_t bound_priority ( uint32_t  prio_in, struct GSF_ConnectedPeer *  cp  )

static

We've received a request with the specified priority.

Bound it according to how much we respect the given peer.

Parameters

prio_in	requested priority
cp	the peer making the request

Returns

effective priority

Definition at line 869 of file gnunet-service-fs_cp.c.

{
#define N ((double) 128.0)
  uint32_t ret;
  double rret;
  int ld;
 
  ld = GSF_test_get_load_too_high_ (0);
  if (GNUNET_SYSERR == ld)
  {
#if INSANE_STATISTICS
    GNUNET_STATISTICS_update (GSF_stats,
                              gettext_noop
                                ("# requests done for free (low load)"), 1,
                              GNUNET_NO);
#endif
    return 0;                   /* excess resources */
  }
  if (prio_in > INT32_MAX)
    prio_in = INT32_MAX;
  ret = -change_peer_respect (cp, -(int) prio_in);
  if (ret > 0)
  {
    if (ret > GSF_current_priorities + N)
      rret = GSF_current_priorities + N;
    else
      rret = ret;
    GSF_current_priorities = (GSF_current_priorities * (N - 1) + rret) / N;
  }
  if ((GNUNET_YES == ld) && (ret > 0))
  {
    /* try with charging */
    ld = GSF_test_get_load_too_high_ (ret);
  }
  if (GNUNET_YES == ld)
  {
    GNUNET_STATISTICS_update (GSF_stats,
                              gettext_noop
                                ("# request dropped, priority insufficient"), 1,
                              GNUNET_NO);
    /* undo charge */
    change_peer_respect (cp, (int) ret);
    return -1;                  /* not enough resources */
  }
  else
  {
    GNUNET_STATISTICS_update (GSF_stats,
                              gettext_noop
                                ("# requests done for a price (normal load)"),
                              1,
                              GNUNET_NO);
  }
#undef N
  return ret;
}

References change_peer_respect(), GSF_DelayedHandle::cp, gettext_noop, GNUNET_NO, GNUNET_STATISTICS_update(), GNUNET_SYSERR, GNUNET_YES, GSF_current_priorities, GSF_stats, GSF_test_get_load_too_high_(), N, and ret.

Referenced by handle_p2p_get().

Here is the call graph for this function:

Here is the caller graph for this function:

bound_ttl()

static int32_t bound_ttl ( int32_t  ttl_in, uint32_t  prio  )

static

The priority level imposes a bound on the maximum value for the ttl that can be requested.

Parameters

ttl_in	requested ttl
prio	given priority

Returns

ttl_in if ttl_in is below the limit, otherwise the ttl-limit for the given prio

Definition at line 937 of file gnunet-service-fs_cp.c.

{
  unsigned long long allowed;
 
  if (ttl_in <= 0)
    return ttl_in;
  allowed = ((unsigned long long) prio) * TTL_DECREMENT / 1000;
  if (ttl_in > allowed)
  {
    if (allowed >= (1 << 30))
      return 1 << 30;
    return allowed;
  }
  return ttl_in;
}

References TTL_DECREMENT.

Referenced by handle_p2p_get().

Here is the caller graph for this function:

test_exist_cb()

static int test_exist_cb ( void *  cls, const struct GNUNET_HashCode *  hc, void *  value  )

static

Test if the query already exists.

If so, merge it, otherwise keep finished at GNUNET_NO.

Parameters

cls	our struct TestExistClosure
hc	the key of the query
value	the existing struct PeerRequest.

Returns

GNUNET_YES to continue to iterate, GNUNET_NO if we successfully merged

Definition at line 993 of file gnunet-service-fs_cp.c.

{
  struct TestExistClosure *tec = cls;
  struct PeerRequest *peerreq = value;
  struct GSF_PendingRequest *pr;
  struct GSF_PendingRequestData *prd;
 
  pr = peerreq->pr;
  prd = GSF_pending_request_get_data_ (pr);
  if (prd->type != tec->type)
    return GNUNET_YES;
  if (prd->ttl.abs_value_us >=
      GNUNET_TIME_absolute_get ().abs_value_us + tec->ttl * 1000LL)
  {
    /* existing request has higher TTL, drop new one! */
    prd->priority += tec->priority;
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Have existing request with higher TTL, dropping new request.\n");
    GNUNET_STATISTICS_update (GSF_stats,
                              gettext_noop
                                ("# requests dropped due to higher-TTL request")
                              ,
                              1, GNUNET_NO);
    tec->finished = GNUNET_YES;
    return GNUNET_NO;
  }
  /* existing request has lower TTL, drop old one! */
  tec->priority += prd->priority;
  free_pending_request (peerreq);
  GSF_pending_request_cancel_ (pr,
                               GNUNET_YES);
  return GNUNET_NO;
}

References GNUNET_TIME_Absolute::abs_value_us, TestExistClosure::finished, free_pending_request(), gettext_noop, GNUNET_ERROR_TYPE_DEBUG, GNUNET_log, GNUNET_NO, GNUNET_STATISTICS_update(), GNUNET_TIME_absolute_get(), GNUNET_YES, GSF_pending_request_cancel_(), GSF_pending_request_get_data_(), GSF_stats, PeerRequest::pr, TestExistClosure::priority, GSF_PendingRequestData::priority, TestExistClosure::ttl, GSF_PendingRequestData::ttl, TestExistClosure::type, GSF_PendingRequestData::type, and value.

Referenced by handle_p2p_get().

Here is the call graph for this function:

Here is the caller graph for this function:

handle_p2p_get()

void handle_p2p_get	(	void *	cls,
		const struct GetMessage *	gm
	)

Handle P2P "QUERY" message.

Creates the pending request entry and sets up all of the data structures to that we will process replies properly. Does not initiate forwarding or local database lookups.

Parameters

cls	the other peer involved (sender of the message)
gm	the GET message

Definition at line 1040 of file gnunet-service-fs_cp.c.

{
  struct GSF_ConnectedPeer *cps = cls;
  struct PeerRequest *peerreq;
  struct GSF_PendingRequest *pr;
  struct GSF_ConnectedPeer *cp;
  const struct GNUNET_PeerIdentity *target;
  enum GSF_PendingRequestOptions options;
  uint16_t msize;
  unsigned int bits;
  const struct GNUNET_PeerIdentity *opt;
  uint32_t bm;
  size_t bfsize;
  uint32_t ttl_decrement;
  struct TestExistClosure tec;
  GNUNET_PEER_Id spid;
  const struct GSF_PendingRequestData *prd;
 
  msize = ntohs (gm->header.size);
  tec.type = ntohl (gm->type);
  bm = ntohl (gm->hash_bitmap);
  bits = 0;
  while (bm > 0)
  {
    if (1 == (bm & 1))
      bits++;
    bm >>= 1;
  }
  opt = (const struct GNUNET_PeerIdentity *) &gm[1];
  bfsize = msize - sizeof(struct GetMessage) - bits * sizeof(struct
                                                             GNUNET_PeerIdentity);
  GNUNET_STATISTICS_update (GSF_stats,
                            gettext_noop
                              ("# GET requests received (from other peers)"),
                            1,
                            GNUNET_NO);
  GSF_cover_query_count++;
  bm = ntohl (gm->hash_bitmap);
  bits = 0;
  if (0 != (bm & GET_MESSAGE_BIT_RETURN_TO))
    cp = GSF_peer_get_ (&opt[bits++]);
  else
    cp = cps;
  if (NULL == cp)
  {
    if (0 != (bm & GET_MESSAGE_BIT_RETURN_TO))
      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                  "Failed to find RETURN-TO peer `%s' in connection set. Dropping query.\n",
                  GNUNET_i2s (&opt[bits - 1]));
 
    else
      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                  "Failed to find peer `%s' in connection set. Dropping query.\n",
                  GNUNET_i2s (cps->ppd.peer));
    GNUNET_STATISTICS_update (GSF_stats,
                              gettext_noop
                              (
                                "# requests dropped due to missing reverse route"),
                              1,
                              GNUNET_NO);
    return;
  }
  unsigned int queue_size = GNUNET_MQ_get_length (cp->mq);
  queue_size += cp->ppd.pending_replies + cp->delay_queue_size;
  if (queue_size > MAX_QUEUE_PER_PEER)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Peer `%s' has too many replies queued already. Dropping query.\n",
                GNUNET_i2s (cps->ppd.peer));
    GNUNET_STATISTICS_update (GSF_stats,
                              gettext_noop (
                                "# requests dropped due to full reply queue"),
                              1,
                              GNUNET_NO);
    return;
  }
  /* note that we can really only check load here since otherwise
   * peers could find out that we are overloaded by not being
   * disconnected after sending us a malformed query... */
  tec.priority = bound_priority (ntohl (gm->priority),
                                 cps);
  if (tec.priority < 0)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Dropping query from `%s', this peer is too busy.\n",
                GNUNET_i2s (cps->ppd.peer));
    return;
  }
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Received request for `%s' of type %u from peer `%s' with flags %u\n",
              GNUNET_h2s (&gm->query),
              (unsigned int) tec.type,
              GNUNET_i2s (cps->ppd.peer),
              (unsigned int) bm);
  target =
    (0 !=
     (bm & GET_MESSAGE_BIT_TRANSMIT_TO)) ? (&opt[bits++]) : NULL;
  options = GSF_PRO_DEFAULTS;
  spid = 0;
  if ((GNUNET_LOAD_get_load (cp->ppd.transmission_delay) > 3 * (1
                                                                + tec.priority))
      || (GNUNET_LOAD_get_average (cp->ppd.transmission_delay) >
          GNUNET_CONSTANTS_MAX_CORK_DELAY.rel_value_us * 2
          + GNUNET_LOAD_get_average (GSF_rt_entry_lifetime)))
  {
    /* don't have BW to send to peer, or would likely take longer than we have for it,
     * so at best indirect the query */
    tec.priority = 0;
    options |= GSF_PRO_FORWARD_ONLY;
    spid = GNUNET_PEER_intern (cps->ppd.peer);
    GNUNET_assert (0 != spid);
  }
  tec.ttl = bound_ttl (ntohl (gm->ttl),
                       tec.priority);
  /* decrement ttl (always) */
  ttl_decrement =
    2 * TTL_DECREMENT + GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
                                                  TTL_DECREMENT);
  if ((tec.ttl < 0) &&
      (((int32_t) (tec.ttl - ttl_decrement)) > 0))
  {
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Dropping query from `%s' due to TTL underflow (%d - %u).\n",
                GNUNET_i2s (cps->ppd.peer),
                tec.ttl,
                ttl_decrement);
    GNUNET_STATISTICS_update (GSF_stats,
                              gettext_noop
                                ("# requests dropped due TTL underflow"), 1,
                              GNUNET_NO);
    /* integer underflow => drop (should be very rare)! */
    return;
  }
  tec.ttl -= ttl_decrement;
 
  /* test if the request already exists */
  tec.finished = GNUNET_NO;
  GNUNET_CONTAINER_multihashmap_get_multiple (cp->request_map,
                                              &gm->query,
                                              &test_exist_cb,
                                              &tec);
  if (GNUNET_YES == tec.finished)
    return; /* merged into existing request, we're done */
 
  peerreq = GNUNET_new (struct PeerRequest);
  peerreq->cp = cp;
  pr = GSF_pending_request_create_ (options,
                                    tec.type,
                                    &gm->query,
                                    target,
                                    (bfsize > 0)
                                    ? (const char *) &opt[bits]
                                    : NULL,
                                    bfsize,
                                    1 /* anonymity */,
                                    (uint32_t) tec.priority,
                                    tec.ttl,
                                    spid,
                                    GNUNET_PEER_intern (cps->ppd.peer),
                                    NULL, 0,        /* replies_seen */
                                    &handle_p2p_reply,
                                    peerreq);
  GNUNET_assert (NULL != pr);
  prd = GSF_pending_request_get_data_ (pr);
  peerreq->pr = pr;
  GNUNET_break (GNUNET_OK ==
                GNUNET_CONTAINER_multihashmap_put (cp->request_map,
                                                   &prd->query,
                                                   peerreq,
                                                   GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE));
  GNUNET_STATISTICS_update (GSF_stats,
                            gettext_noop (
                              "# P2P query messages received and processed"),
                            1,
                            GNUNET_NO);
  GNUNET_STATISTICS_update (GSF_stats,
                            gettext_noop ("# P2P searches active"),
                            1,
                            GNUNET_NO);
  GSF_pending_request_get_data_ (pr)->has_started = GNUNET_YES;
  GSF_local_lookup_ (pr,
                     &GSF_consider_forwarding,
                     NULL);
}

References bound_priority(), bound_ttl(), PeerRequest::cp, GSF_ConnectedPeer::delay_queue_size, TestExistClosure::finished, GET_MESSAGE_BIT_RETURN_TO, GET_MESSAGE_BIT_TRANSMIT_TO, gettext_noop, GNUNET_assert, GNUNET_break, GNUNET_CONSTANTS_MAX_CORK_DELAY, GNUNET_CONTAINER_multihashmap_get_multiple(), GNUNET_CONTAINER_multihashmap_put(), GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE, GNUNET_CRYPTO_QUALITY_WEAK, GNUNET_CRYPTO_random_u32(), GNUNET_ERROR_TYPE_DEBUG, GNUNET_h2s(), GNUNET_i2s(), GNUNET_LOAD_get_average(), GNUNET_LOAD_get_load(), GNUNET_log, GNUNET_MQ_get_length(), GNUNET_new, GNUNET_NO, GNUNET_OK, GNUNET_PEER_intern(), GNUNET_STATISTICS_update(), GNUNET_YES, GSF_consider_forwarding(), GSF_cover_query_count, GSF_local_lookup_(), GSF_peer_get_(), GSF_pending_request_create_(), GSF_pending_request_get_data_(), GSF_PRO_DEFAULTS, GSF_PRO_FORWARD_ONLY, GSF_rt_entry_lifetime, GSF_stats, handle_p2p_reply(), GSF_PendingRequestData::has_started, GetMessage::hash_bitmap, GetMessage::header, MAX_QUEUE_PER_PEER, GSF_ConnectedPeer::mq, options, GSF_PeerPerformanceData::peer, GSF_PeerPerformanceData::pending_replies, GSF_ConnectedPeer::ppd, PeerRequest::pr, GetMessage::priority, TestExistClosure::priority, GetMessage::query, GSF_PendingRequestData::query, GSF_ConnectedPeer::request_map, GNUNET_MessageHeader::size, test_exist_cb(), GSF_PeerPerformanceData::transmission_delay, GetMessage::ttl, TestExistClosure::ttl, TTL_DECREMENT, GetMessage::type, and TestExistClosure::type.

Here is the call graph for this function:

GSF_peer_transmit_()

void GSF_peer_transmit_	(	struct GSF_ConnectedPeer *	cp,
		int	is_query,
		uint32_t	priority,
		struct GNUNET_MQ_Envelope *	env
	)

Transmit a message to the given peer as soon as possible.

If the peer disconnects before the transmission can happen, the callback is invoked with a NULL buffer.

Parameters

cp	target peer
is_query	is this a query (GNUNET_YES) or content (GNUNET_NO) or neither (GNUNET_SYSERR)
priority	how important is this request?
env	message to send

Definition at line 1238 of file gnunet-service-fs_cp.c.

{
  struct GSF_PeerTransmitHandle *pth;
  struct GSF_PeerTransmitHandle *pos;
  struct GSF_PeerTransmitHandle *prev;
 
  pth = GNUNET_new (struct GSF_PeerTransmitHandle);
  pth->transmission_request_start_time = GNUNET_TIME_absolute_get ();
  pth->env = env;
  pth->is_query = is_query;
  pth->priority = priority;
  pth->cp = cp;
  /* insertion sort (by priority, descending) */
  prev = NULL;
  pos = cp->pth_head;
  while ((NULL != pos) && (pos->priority > priority))
  {
    prev = pos;
    pos = pos->next;
  }
  GNUNET_CONTAINER_DLL_insert_after (cp->pth_head,
                                     cp->pth_tail,
                                     prev,
                                     pth);
  if (GNUNET_YES == is_query)
    cp->ppd.pending_queries++;
  else if (GNUNET_NO == is_query)
    cp->ppd.pending_replies++;
  schedule_transmission (pth);
}

References GSF_PeerTransmitHandle::cp, env, GSF_PeerTransmitHandle::env, GNUNET_CONTAINER_DLL_insert_after, GNUNET_new, GNUNET_NO, GNUNET_TIME_absolute_get(), GNUNET_YES, GSF_PeerTransmitHandle::is_query, GSF_PeerTransmitHandle::next, GSF_PeerPerformanceData::pending_queries, GSF_PeerPerformanceData::pending_replies, GSF_ConnectedPeer::ppd, GSF_PeerTransmitHandle::prev, GSF_PeerTransmitHandle::priority, GSF_ConnectedPeer::pth_head, GSF_ConnectedPeer::pth_tail, schedule_transmission(), and GSF_PeerTransmitHandle::transmission_request_start_time.

Referenced by GSF_block_peer_migration_(), handle_p2p_reply(), schedule_peer_transmission(), transmit_content(), and transmit_delayed_now().

Here is the call graph for this function:

Here is the caller graph for this function:

GSF_peer_update_performance_()

void GSF_peer_update_performance_	(	struct GSF_ConnectedPeer *	cp,
		struct GNUNET_TIME_Absolute	request_time,
		uint32_t	request_priority
	)

Report on receiving a reply; update the performance record of the given peer.

Parameters

cp	responding peer (will be updated)
request_time	time at which the original query was transmitted
request_priority	priority of the original request

Definition at line 1281 of file gnunet-service-fs_cp.c.

{
  struct GNUNET_TIME_Relative delay;
 
  delay = GNUNET_TIME_absolute_get_duration (request_time);
  cp->ppd.avg_reply_delay.rel_value_us =
    (cp->ppd.avg_reply_delay.rel_value_us * (RUNAVG_DELAY_N - 1)
     + delay.rel_value_us) / RUNAVG_DELAY_N;
  cp->ppd.avg_priority =
    (cp->ppd.avg_priority * (RUNAVG_DELAY_N - 1)
     + request_priority) / RUNAVG_DELAY_N;
}

References GSF_PeerPerformanceData::avg_priority, GSF_PeerPerformanceData::avg_reply_delay, GNUNET_TIME_absolute_get_duration(), GSF_ConnectedPeer::ppd, GNUNET_TIME_Relative::rel_value_us, and RUNAVG_DELAY_N.

Referenced by update_request_performance_data().

Here is the call graph for this function:

Here is the caller graph for this function:

GSF_peer_update_responder_client_()

void GSF_peer_update_responder_client_	(	struct GSF_ConnectedPeer *	cp,
		struct GSF_LocalClient *	initiator_client
	)

Report on receiving a reply in response to an initiating client.

Remember that this peer is good for this client.

Parameters

cp	responding peer (will be updated)
initiator_client	local client on responsible for query

Definition at line 1305 of file gnunet-service-fs_cp.c.

{
  cp->ppd.last_client_replies[cp->last_client_replies_woff++
                              % CS2P_SUCCESS_LIST_SIZE] = initiator_client;
}

References CS2P_SUCCESS_LIST_SIZE, GSF_PeerPerformanceData::last_client_replies, GSF_ConnectedPeer::last_client_replies_woff, and GSF_ConnectedPeer::ppd.

GSF_peer_update_responder_peer_()

void GSF_peer_update_responder_peer_	(	struct GSF_ConnectedPeer *	cp,
		const struct GSF_ConnectedPeer *	initiator_peer
	)

Report on receiving a reply in response to an initiating peer.

Remember that this peer is good for this initiating peer.

Parameters

cp	responding peer (will be updated)
initiator_peer	other peer responsible for query

Definition at line 1321 of file gnunet-service-fs_cp.c.

{
  unsigned int woff;
 
  woff = cp->last_p2p_replies_woff % P2P_SUCCESS_LIST_SIZE;
  GNUNET_PEER_change_rc (cp->ppd.last_p2p_replies[woff], -1);
  cp->ppd.last_p2p_replies[woff] = initiator_peer->ppd.pid;
  GNUNET_PEER_change_rc (initiator_peer->ppd.pid, 1);
  cp->last_p2p_replies_woff = (woff + 1) % P2P_SUCCESS_LIST_SIZE;
}

References GNUNET_PEER_change_rc(), GSF_PeerPerformanceData::last_p2p_replies, GSF_ConnectedPeer::last_p2p_replies_woff, P2P_SUCCESS_LIST_SIZE, GSF_PeerPerformanceData::pid, and GSF_ConnectedPeer::ppd.

Here is the call graph for this function:

flush_respect()

static int flush_respect ( void *  cls, const struct GNUNET_PeerIdentity *  key, void *  value  )

static

Write peer-respect information to a file - flush the buffer entry!

Parameters

cls	unused
key	peer identity
value	the struct GSF_ConnectedPeer to flush

Returns

GNUNET_OK to continue iteration

Definition at line 1343 of file gnunet-service-fs_cp.c.

{
  struct GSF_ConnectedPeer *cp = value;
  struct GNUNET_PeerIdentity pid;
 
  if (cp->ppd.respect == cp->disk_respect)
    return GNUNET_OK;           /* unchanged */
  GNUNET_assert (0 != cp->ppd.pid);
  GNUNET_PEER_resolve (cp->ppd.pid, &pid);
  GNUNET_PEERSTORE_store (peerstore, "fs", &pid, "respect", &cp->ppd.respect,
                          sizeof(cp->ppd.respect),
                          GNUNET_TIME_UNIT_FOREVER_ABS,
                          GNUNET_PEERSTORE_STOREOPTION_REPLACE,
                          NULL,
                          NULL);
  return GNUNET_OK;
}

References GSF_ConnectedPeer::disk_respect, GNUNET_assert, GNUNET_OK, GNUNET_PEER_resolve(), GNUNET_PEERSTORE_store(), GNUNET_PEERSTORE_STOREOPTION_REPLACE, GNUNET_TIME_UNIT_FOREVER_ABS, peerstore, GSF_PeerPerformanceData::pid, pid, GSF_ConnectedPeer::ppd, GSF_PeerPerformanceData::respect, and value.

Referenced by cron_flush_respect(), GSF_connected_peer_done_(), and GSF_peer_disconnect_handler().

Here is the call graph for this function:

Here is the caller graph for this function:

GSF_peer_disconnect_handler()

void GSF_peer_disconnect_handler	(	void *	cls,
		const struct GNUNET_PeerIdentity *	peer,
		void *	internal_cls
	)

A peer disconnected from us.

Tear down the connected peer record.

Parameters

cls	unused
peer	identity of peer that disconnected
internal_cls	the corresponding struct GSF_ConnectedPeer

Definition at line 1365 of file gnunet-service-fs_cp.c.

{
  struct GSF_ConnectedPeer *cp = internal_cls;
  struct GSF_PeerTransmitHandle *pth;
  struct GSF_DelayedHandle *dh;
 
  if (NULL == cp)
    return;  /* must have been disconnect from core with
              * 'peer' == my_id, ignore */
  flush_respect (NULL,
                 peer,
                 cp);
  GNUNET_assert (GNUNET_YES ==
                 GNUNET_CONTAINER_multipeermap_remove (cp_map,
                                                       peer,
                                                       cp));
  GNUNET_STATISTICS_set (GSF_stats,
                         gettext_noop ("# peers connected"),
                         GNUNET_CONTAINER_multipeermap_size (cp_map),
                         GNUNET_NO);
  if (NULL != cp->respect_iterate_req)
  {
    GNUNET_PEERSTORE_iteration_stop (cp->respect_iterate_req);
    cp->respect_iterate_req = NULL;
  }
  GNUNET_CONTAINER_multihashmap_iterate (cp->request_map,
                                         &cancel_pending_request,
                                         cp);
  GNUNET_CONTAINER_multihashmap_destroy (cp->request_map);
  cp->request_map = NULL;
  GSF_plan_notify_peer_disconnect_ (cp);
  GNUNET_LOAD_value_free (cp->ppd.transmission_delay);
  GNUNET_PEER_decrement_rcs (cp->ppd.last_p2p_replies,
                             P2P_SUCCESS_LIST_SIZE);
  memset (cp->ppd.last_p2p_replies,
          0,
          sizeof(cp->ppd.last_p2p_replies));
  GSF_push_stop_ (cp);
  while (NULL != (pth = cp->pth_head))
  {
    GNUNET_CONTAINER_DLL_remove (cp->pth_head,
                                 cp->pth_tail,
                                 pth);
    if (GNUNET_YES == pth->is_query)
      GNUNET_assert (0 < cp->ppd.pending_queries--);
    else if (GNUNET_NO == pth->is_query)
      GNUNET_assert (0 < cp->ppd.pending_replies--);
    GNUNET_free (pth);
  }
  while (NULL != (dh = cp->delayed_head))
  {
    GNUNET_CONTAINER_DLL_remove (cp->delayed_head,
                                 cp->delayed_tail,
                                 dh);
    GNUNET_MQ_discard (dh->env);
    cp->delay_queue_size--;
    GNUNET_SCHEDULER_cancel (dh->delay_task);
    GNUNET_free (dh);
  }
  GNUNET_PEER_change_rc (cp->ppd.pid, -1);
  if (NULL != cp->mig_revive_task)
  {
    GNUNET_SCHEDULER_cancel (cp->mig_revive_task);
    cp->mig_revive_task = NULL;
  }
  GNUNET_break (0 == cp->ppd.pending_queries);
  GNUNET_break (0 == cp->ppd.pending_replies);
  GNUNET_free (cp);
}

References cancel_pending_request(), GSF_DelayedHandle::cp, cp_map, GSF_ConnectedPeer::delay_queue_size, GSF_DelayedHandle::delay_task, GSF_ConnectedPeer::delayed_head, GSF_ConnectedPeer::delayed_tail, GSF_DelayedHandle::env, flush_respect(), gettext_noop, GNUNET_assert, GNUNET_break, GNUNET_CONTAINER_DLL_remove, GNUNET_CONTAINER_multihashmap_destroy(), GNUNET_CONTAINER_multihashmap_iterate(), GNUNET_CONTAINER_multipeermap_remove(), GNUNET_CONTAINER_multipeermap_size(), GNUNET_free, GNUNET_LOAD_value_free, GNUNET_MQ_discard(), GNUNET_NO, GNUNET_PEER_change_rc(), GNUNET_PEER_decrement_rcs(), GNUNET_PEERSTORE_iteration_stop(), GNUNET_SCHEDULER_cancel(), GNUNET_STATISTICS_set(), GNUNET_YES, GSF_plan_notify_peer_disconnect_(), GSF_push_stop_(), GSF_stats, GSF_PeerTransmitHandle::is_query, GSF_PeerPerformanceData::last_p2p_replies, GSF_ConnectedPeer::mig_revive_task, P2P_SUCCESS_LIST_SIZE, GSF_PeerPerformanceData::pending_queries, GSF_PeerPerformanceData::pending_replies, GSF_PeerPerformanceData::pid, GSF_ConnectedPeer::ppd, GSF_ConnectedPeer::pth_head, GSF_ConnectedPeer::pth_tail, GSF_ConnectedPeer::request_map, GSF_ConnectedPeer::respect_iterate_req, and GSF_PeerPerformanceData::transmission_delay.

Referenced by main_init().

Here is the call graph for this function:

Here is the caller graph for this function:

call_iterator()

static int call_iterator ( void *  cls, const struct GNUNET_PeerIdentity *  key, void *  value  )

static

Function that calls the callback for each peer.

Parameters

cls	the struct IterationContext *
key	identity of the peer
value	the struct GSF_ConnectedPeer *

Returns

GNUNET_YES to continue iteration

Definition at line 1464 of file gnunet-service-fs_cp.c.

{
  struct IterationContext *ic = cls;
  struct GSF_ConnectedPeer *cp = value;
 
  ic->it (ic->it_cls,
          key, cp,
          &cp->ppd);
  return GNUNET_YES;
}

References GNUNET_YES, IterationContext::it, IterationContext::it_cls, key, GSF_ConnectedPeer::ppd, and value.

Referenced by GSF_iterate_connected_peers_().

Here is the caller graph for this function:

GSF_iterate_connected_peers_()

void GSF_iterate_connected_peers_	(	GSF_ConnectedPeerIterator	it,
		void *	it_cls
	)

Iterate over all connected peers.

Parameters

it	function to call for each peer
it_cls	closure for it

Definition at line 1479 of file gnunet-service-fs_cp.c.

{
  struct IterationContext ic;
 
  ic.it = it;
  ic.it_cls = it_cls;
  GNUNET_CONTAINER_multipeermap_iterate (cp_map,
                                         &call_iterator,
                                         &ic);
}

References call_iterator(), cp_map, GNUNET_CONTAINER_multipeermap_iterate(), IterationContext::it, and IterationContext::it_cls.

Referenced by GSF_consider_forwarding().

Here is the call graph for this function:

Here is the caller graph for this function:

GSF_connected_peer_get_identity_()

void GSF_connected_peer_get_identity_	(	const struct GSF_ConnectedPeer *	cp,
		struct GNUNET_PeerIdentity *	id
	)

Obtain the identity of a connected peer.

Parameters

cp	peer to get identity of
id	identity to set (written to)

Definition at line 1499 of file gnunet-service-fs_cp.c.

{
  GNUNET_assert (0 != cp->ppd.pid);
  GNUNET_PEER_resolve (cp->ppd.pid, id);
}

References GNUNET_assert, GNUNET_PEER_resolve(), GSF_PeerPerformanceData::pid, and GSF_ConnectedPeer::ppd.

Referenced by consider_peer_for_forwarding().

Here is the call graph for this function:

Here is the caller graph for this function:

GSF_connected_peer_get_identity2_()

const struct GNUNET_PeerIdentity * GSF_connected_peer_get_identity2_

(

const struct GSF_ConnectedPeer *

cp

)

Obtain the identity of a connected peer.

Parameters

cp	peer to get identity of

Returns

reference to peer identity, valid until peer disconnects (!)

Definition at line 1514 of file gnunet-service-fs_cp.c.

{
  GNUNET_assert (0 != cp->ppd.pid);
  return GNUNET_PEER_resolve2 (cp->ppd.pid);
}

References GNUNET_assert, GNUNET_PEER_resolve2(), GSF_PeerPerformanceData::pid, and GSF_ConnectedPeer::ppd.

Referenced by GSF_peer_connect_handler(), GSF_plan_add_(), GSF_plan_notify_peer_disconnect_(), GSF_push_start_(), and process_migration_content().

Here is the call graph for this function:

Here is the caller graph for this function:

GSF_block_peer_migration_()

void GSF_block_peer_migration_	(	struct GSF_ConnectedPeer *	cp,
		struct GNUNET_TIME_Absolute	block_time
	)

Ask a peer to stop migrating data to us until the given point in time.

Parameters

cp	peer to ask
block_time	until when to block

Definition at line 1529 of file gnunet-service-fs_cp.c.

{
  struct GNUNET_MQ_Envelope *env;
  struct MigrationStopMessage *msm;
 
  if (cp->last_migration_block.abs_value_us > block_time.abs_value_us)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Migration already blocked for another %s\n",
                GNUNET_STRINGS_relative_time_to_string (
                  GNUNET_TIME_absolute_get_remaining
                    (cp->
                    last_migration_block), GNUNET_YES));
    return;                     /* already blocked */
  }
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Asking to stop migration for %s\n",
              GNUNET_STRINGS_relative_time_to_string (
                GNUNET_TIME_absolute_get_remaining (block_time),
                GNUNET_YES));
  cp->last_migration_block = block_time;
  env = GNUNET_MQ_msg (msm,
                       GNUNET_MESSAGE_TYPE_FS_MIGRATION_STOP);
  msm->reserved = htonl (0);
  msm->duration
    = GNUNET_TIME_relative_hton (GNUNET_TIME_absolute_get_remaining
                                   (cp->last_migration_block));
  GNUNET_STATISTICS_update (GSF_stats,
                            gettext_noop ("# migration stop messages sent"),
                            1,
                            GNUNET_NO);
  GSF_peer_transmit_ (cp,
                      GNUNET_SYSERR,
                      UINT32_MAX,
                      env);
}

References GNUNET_TIME_Absolute::abs_value_us, MigrationStopMessage::duration, env, gettext_noop, GNUNET_ERROR_TYPE_DEBUG, GNUNET_log, GNUNET_MESSAGE_TYPE_FS_MIGRATION_STOP, GNUNET_MQ_msg, GNUNET_NO, GNUNET_STATISTICS_update(), GNUNET_STRINGS_relative_time_to_string(), GNUNET_SYSERR, GNUNET_TIME_absolute_get_remaining(), GNUNET_TIME_relative_hton(), GNUNET_YES, GSF_peer_transmit_(), GSF_stats, GSF_ConnectedPeer::last_migration_block, and MigrationStopMessage::reserved.

Referenced by handle_p2p_put(), and put_migration_continuation().

Here is the call graph for this function:

Here is the caller graph for this function:

GSF_connected_peer_change_preference_()

void GSF_connected_peer_change_preference_	(	struct GSF_ConnectedPeer *	cp,
		uint64_t	pref
	)

Notify core about a preference we have for the given peer (to allocate more resources towards it).

The change will be communicated the next time we reserve bandwidth with core (not instantly).

Parameters

cp	peer to reserve bandwidth from
pref	preference change

Definition at line 1577 of file gnunet-service-fs_cp.c.

{
  cp->inc_preference += pref;
}

References GSF_ConnectedPeer::inc_preference.

Referenced by handle_p2p_put().

Here is the caller graph for this function:

cron_flush_respect()

static void cron_flush_respect ( void *  cls )

static

Call this method periodically to flush respect information to disk.

Parameters

cls	closure, not used

Definition at line 1590 of file gnunet-service-fs_cp.c.

{
  fr_task = NULL;
  GNUNET_CONTAINER_multipeermap_iterate (cp_map,
                                         &flush_respect,
                                         NULL);
  fr_task = GNUNET_SCHEDULER_add_delayed_with_priority (RESPECT_FLUSH_FREQ,
                                                        GNUNET_SCHEDULER_PRIORITY_HIGH,
                                                        &cron_flush_respect,
                                                        NULL);
}

References cp_map, cron_flush_respect(), flush_respect(), fr_task, GNUNET_CONTAINER_multipeermap_iterate(), GNUNET_SCHEDULER_add_delayed_with_priority(), GNUNET_SCHEDULER_PRIORITY_HIGH, and RESPECT_FLUSH_FREQ.

Referenced by cron_flush_respect(), and GSF_connected_peer_init_().

Here is the call graph for this function:

Here is the caller graph for this function:

GSF_connected_peer_init_()

void GSF_connected_peer_init_

(

void

)

Initialize peer management subsystem.

Definition at line 1607 of file gnunet-service-fs_cp.c.

{
  cp_map = GNUNET_CONTAINER_multipeermap_create (128, GNUNET_YES);
  peerstore = GNUNET_PEERSTORE_connect (GSF_cfg);
  fr_task = GNUNET_SCHEDULER_add_with_priority (GNUNET_SCHEDULER_PRIORITY_HIGH,
                                                &cron_flush_respect, NULL);
}

References cp_map, cron_flush_respect(), fr_task, GNUNET_CONTAINER_multipeermap_create(), GNUNET_PEERSTORE_connect(), GNUNET_SCHEDULER_add_with_priority(), GNUNET_SCHEDULER_PRIORITY_HIGH, GNUNET_YES, GSF_cfg, and peerstore.

Referenced by run().

Here is the call graph for this function:

Here is the caller graph for this function:

GSF_connected_peer_done_()

void GSF_connected_peer_done_

(

void

)

Shutdown peer management subsystem.

Definition at line 1620 of file gnunet-service-fs_cp.c.

{
  GNUNET_CONTAINER_multipeermap_iterate (cp_map,
                                         &flush_respect,
                                         NULL);
  GNUNET_SCHEDULER_cancel (fr_task);
  fr_task = NULL;
  GNUNET_CONTAINER_multipeermap_destroy (cp_map);
  cp_map = NULL;
  GNUNET_PEERSTORE_disconnect (peerstore);
}

References cp_map, flush_respect(), fr_task, GNUNET_CONTAINER_multipeermap_destroy(), GNUNET_CONTAINER_multipeermap_iterate(), GNUNET_PEERSTORE_disconnect(), GNUNET_SCHEDULER_cancel(), and peerstore.

Referenced by shutdown_task().

Here is the call graph for this function:

Here is the caller graph for this function:

clean_local_client()

static int clean_local_client ( void *  cls, const struct GNUNET_PeerIdentity *  key, void *  value  )

static

Iterator to remove references to LC entry.

Parameters

cls	the struct GSF_LocalClient * to look for
key	current key code
value	value in the hash map (peer entry)

Returns

GNUNET_YES (we should continue to iterate)

Definition at line 1642 of file gnunet-service-fs_cp.c.

{
  const struct GSF_LocalClient *lc = cls;
  struct GSF_ConnectedPeer *cp = value;
  unsigned int i;
 
  for (i = 0; i < CS2P_SUCCESS_LIST_SIZE; i++)
    if (cp->ppd.last_client_replies[i] == lc)
      cp->ppd.last_client_replies[i] = NULL;
  return GNUNET_YES;
}

References CS2P_SUCCESS_LIST_SIZE, GNUNET_YES, GSF_PeerPerformanceData::last_client_replies, GSF_ConnectedPeer::ppd, and value.

Referenced by GSF_handle_local_client_disconnect_().

Here is the caller graph for this function:

GSF_handle_local_client_disconnect_()

void GSF_handle_local_client_disconnect_

(

const struct GSF_LocalClient *

lc

)

Notification that a local client disconnected.

Clean up all of our references to the given handle.

Parameters

lc	handle to the local client (henceforth invalid)

Definition at line 1664 of file gnunet-service-fs_cp.c.

{
  if (NULL == cp_map)
    return;                     /* already cleaned up */
  GNUNET_CONTAINER_multipeermap_iterate (cp_map,
                                         &clean_local_client,
                                         (void *) lc);
}

References clean_local_client(), cp_map, and GNUNET_CONTAINER_multipeermap_iterate().

Here is the call graph for this function:

Variable Documentation

cp_map

struct GNUNET_CONTAINER_MultiPeerMap* cp_map

static

Map from peer identities to struct GSF_ConnectPeer entries.

Definition at line 268 of file gnunet-service-fs_cp.c.

Referenced by cron_flush_respect(), GSF_connected_peer_done_(), GSF_connected_peer_init_(), GSF_handle_local_client_disconnect_(), GSF_iterate_connected_peers_(), GSF_peer_connect_handler(), GSF_peer_disconnect_handler(), and GSF_peer_get_().

peerstore

struct GNUNET_PEERSTORE_Handle* peerstore

static

Handle to peerstore service.

Definition at line 273 of file gnunet-service-fs_cp.c.

Referenced by flush_respect(), GSF_connected_peer_done_(), GSF_connected_peer_init_(), and GSF_peer_connect_handler().

fr_task

struct GNUNET_SCHEDULER_Task* fr_task

static

Task used to flush respect values to disk.

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

Referenced by cron_flush_respect(), GSF_connected_peer_done_(), and GSF_connected_peer_init_().