API to push content from our datastore to other peers ('anonymous'-content P2P migration) More...

#include "platform.h"
#include "gnunet-service-fs.h"
#include "gnunet-service-fs_cp.h"
#include "gnunet-service-fs_indexing.h"
#include "gnunet-service-fs_push.h"

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

Go to the source code of this file.

Data Structures
struct	MigrationReadyBlock
	Block that is ready for migration to other peers. More...

struct	MigrationReadyPeer
	Information about a peer waiting for migratable data. More...

Macros
#define	MAX_MIGRATION_QUEUE 8
	Maximum number of blocks we keep in memory for migration. More...

#define	MIGRATION_LIST_SIZE 2
	Blocks are at most migrated to this number of peers plus one, each time they are fetched from the database. More...

#define	MIN_MIGRATION_CONTENT_LIFETIME
	How long must content remain valid for us to consider it for migration? If content will expire too soon, there is clearly no point in pushing it to other peers. More...

Functions
static void	delete_migration_block (struct MigrationReadyBlock *mb)
	Delete the given migration block. More...

static void	find_content (void *cls)
	Find content for migration to this peer. More...

static int	transmit_content (struct MigrationReadyPeer mrp, struct MigrationReadyBlock block)
	Send the given block to the given peer. More...

static unsigned int	count_targets (struct MigrationReadyBlock *block)
	Count the number of peers this block has already been forwarded to. More...

static long	score_content (struct MigrationReadyPeer mrp, struct MigrationReadyBlock block)
	Check if sending this block to this peer would be a good idea. More...

static void	consider_gathering (void)
	If the migration task is not currently running, consider (re)scheduling it with the appropriate delay. More...

static void	gather_migration_blocks (void *cls)
	Task that is run periodically to obtain blocks for content migration. More...

static void	process_migration_content (void cls, const struct GNUNET_HashCode key, size_t size, const void *data, enum GNUNET_BLOCK_Type type, uint32_t priority, uint32_t anonymity, uint32_t replication, struct GNUNET_TIME_Absolute expiration, uint64_t uid)
	Process content offered for migration. More...

void	GSF_push_start_ (struct GSF_ConnectedPeer *peer)
	A peer connected to us. More...

void	GSF_push_stop_ (struct GSF_ConnectedPeer *peer)
	A peer disconnected from us. More...

void	GSF_push_init_ ()
	Setup the module. More...

void	GSF_push_done_ ()
	Shutdown the module. More...

Variables
static struct MigrationReadyBlock *	mig_head
	Head of linked list of blocks that can be migrated. More...

static struct MigrationReadyBlock *	mig_tail
	Tail of linked list of blocks that can be migrated. More...

static struct MigrationReadyPeer *	peer_head
	Head of linked list of peers. More...

static struct MigrationReadyPeer *	peer_tail
	Tail of linked list of peers. More...

static struct GNUNET_DATASTORE_QueueEntry *	mig_qe
	Request to datastore for migration (or NULL). More...

static struct GNUNET_SCHEDULER_Task *	mig_task
	ID of task that collects blocks for migration. More...

static struct GNUNET_TIME_Relative	min_migration_delay
	What is the maximum frequency at which we are allowed to poll the datastore for migration content? More...

static unsigned int	mig_size
	Size of the doubly-linked list of migration blocks. More...

static int	enabled
	Is this module enabled? More...

static int	value_found
	Did we find anything in the datastore? More...

Detailed Description

API to push content from our datastore to other peers ('anonymous'-content P2P migration)

Author: Christian Grothoff

Definition in file gnunet-service-fs_push.c.

Macro Definition Documentation

◆ MAX_MIGRATION_QUEUE

#define MAX_MIGRATION_QUEUE 8

Maximum number of blocks we keep in memory for migration.

Definition at line 37 of file gnunet-service-fs_push.c.

◆ MIGRATION_LIST_SIZE

#define MIGRATION_LIST_SIZE 2

Blocks are at most migrated to this number of peers plus one, each time they are fetched from the database.

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

◆ MIN_MIGRATION_CONTENT_LIFETIME

#define MIN_MIGRATION_CONTENT_LIFETIME

Value:

GNUNET_TIME_relative_multiply ( \

GNUNET_TIME_UNIT_MINUTES, 30)

GNUNET_TIME_UNIT_MINUTES

#define GNUNET_TIME_UNIT_MINUTES

One minute.

Definition: gnunet_time_lib.h:161

GNUNET_TIME_relative_multiply

struct GNUNET_TIME_Relative GNUNET_TIME_relative_multiply(struct GNUNET_TIME_Relative rel, unsigned long long factor)

Multiply relative time by a given factor.

Definition: time.c:486

How long must content remain valid for us to consider it for migration? If content will expire too soon, there is clearly no point in pushing it to other peers.

This value gives the threshold for migration. Note that if this value is increased, the migration testcase may need to be adjusted as well (especially the CONTENT_LIFETIME in fs_test_lib.c).

Definition at line 52 of file gnunet-service-fs_push.c.

Function Documentation

◆ delete_migration_block()

static void delete_migration_block ( struct MigrationReadyBlock * mb )

static

Delete the given migration block.

Parameters

mb	block to delete

Definition at line 189 of file gnunet-service-fs_push.c.

{
  GNUNET_CONTAINER_DLL_remove (mig_head,
                               mig_tail,
                               mb);
  GNUNET_PEER_decrement_rcs (mb->target_list,
                             MIGRATION_LIST_SIZE);
  mig_size--;
  GNUNET_free (mb);
}

References GNUNET_CONTAINER_DLL_remove, GNUNET_free, GNUNET_PEER_decrement_rcs(), mig_head, mig_size, mig_tail, MIGRATION_LIST_SIZE, and MigrationReadyBlock::target_list.

Referenced by find_content(), GSF_push_done_(), and transmit_content().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ find_content()

static void find_content ( void * cls )

static

Find content for migration to this peer.

Parameters

cls	A `struct MigrationReadyPeer *` to find content for

Definition at line 329 of file gnunet-service-fs_push.c.

{
  struct MigrationReadyPeer *mrp = cls;
  struct MigrationReadyBlock *pos;
  long score;
  long best_score;
  struct MigrationReadyBlock *best;
 
  mrp->env = NULL;
  best = NULL;
  best_score = -1;
  pos = mig_head;
  while (NULL != pos)
  {
    score = score_content (mrp, pos);
    if (score > best_score)
    {
      best_score = score;
      best = pos;
    }
    pos = pos->next;
  }
  if (NULL == best)
  {
    if (mig_size < MAX_MIGRATION_QUEUE)
    {
      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                  "No content found for pushing, waiting for queue to fill\n");
      return;                   /* will fill up eventually... */
    }
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "No suitable content found, purging content from full queue\n");
    /* failed to find migration target AND
     * queue is full, purge most-forwarded
     * block from queue to make room for more */
    pos = mig_head;
    while (NULL != pos)
    {
      score = count_targets (pos);
      if (score >= best_score)
      {
        best_score = score;
        best = pos;
      }
      pos = pos->next;
    }
    GNUNET_assert (NULL != best);
    delete_migration_block (best);
    consider_gathering ();
    return;
  }
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Preparing to push best content to peer\n");
  transmit_content (mrp,
                    best);
}

References consider_gathering(), count_targets(), delete_migration_block(), MigrationReadyPeer::env, GNUNET_assert, GNUNET_ERROR_TYPE_DEBUG, GNUNET_log, MAX_MIGRATION_QUEUE, mig_head, mig_size, MigrationReadyBlock::next, score_content(), and transmit_content().

Referenced by GSF_push_start_(), and transmit_content().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ transmit_content()

static int transmit_content	(	struct MigrationReadyPeer *	mrp,
		struct MigrationReadyBlock *	block
	)

static

Send the given block to the given peer.

Parameters

mrp	target peer
block	the block

Returns: GNUNET_YES if the block was deleted (!)

Definition at line 218 of file gnunet-service-fs_push.c.

{
  struct PutMessage *msg;
  unsigned int i;
  struct GSF_PeerPerformanceData *ppd;
  int ret;
 
  ppd = GSF_get_peer_performance_data_ (mrp->peer);
  GNUNET_assert (NULL == mrp->env);
  mrp->env = GNUNET_MQ_msg_extra (msg,
                                  block->size,
                                  GNUNET_MESSAGE_TYPE_FS_PUT);
  msg->type = htonl (block->type);
  msg->expiration = GNUNET_TIME_absolute_hton (block->expiration);
  GNUNET_memcpy (&msg[1],
                 &block[1],
                 block->size);
  for (i = 0; i < MIGRATION_LIST_SIZE; i++)
  {
    if (block->target_list[i] == 0)
    {
      block->target_list[i] = ppd->pid;
      GNUNET_PEER_change_rc (block->target_list[i],
                             1);
      break;
    }
  }
  if (MIGRATION_LIST_SIZE == i)
  {
    delete_migration_block (block);
    ret = GNUNET_YES;
  }
  else
  {
    ret = GNUNET_NO;
  }
  GNUNET_MQ_notify_sent (mrp->env,
                         &find_content,
                         mrp);
  GSF_peer_transmit_ (mrp->peer,
                      GNUNET_NO,
                      0 /* priority */,
                      mrp->env);
  return ret;
}

References delete_migration_block(), MigrationReadyPeer::env, MigrationReadyBlock::expiration, find_content(), GNUNET_assert, GNUNET_memcpy, GNUNET_MESSAGE_TYPE_FS_PUT, GNUNET_MQ_msg_extra, GNUNET_MQ_notify_sent(), GNUNET_NO, GNUNET_PEER_change_rc(), GNUNET_TIME_absolute_hton(), GNUNET_YES, GSF_get_peer_performance_data_(), GSF_peer_transmit_(), MIGRATION_LIST_SIZE, msg, MigrationReadyPeer::peer, GSF_PeerPerformanceData::pid, ret, MigrationReadyBlock::size, MigrationReadyBlock::target_list, GNUNET_MessageHeader::type, and MigrationReadyBlock::type.

Referenced by find_content(), and process_migration_content().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ count_targets()

static unsigned int count_targets ( struct MigrationReadyBlock * block )

static

Count the number of peers this block has already been forwarded to.

Parameters

block the block

Returns: number of times block was forwarded

Definition at line 274 of file gnunet-service-fs_push.c.

{
  unsigned int i;
 
  for (i = 0; i < MIGRATION_LIST_SIZE; i++)
    if (block->target_list[i] == 0)
      return i;
  return i;
}

References MIGRATION_LIST_SIZE, and MigrationReadyBlock::target_list.

Referenced by find_content().

Here is the caller graph for this function:

◆ score_content()

static long score_content	(	struct MigrationReadyPeer *	mrp,
		struct MigrationReadyBlock *	block
	)

static

Check if sending this block to this peer would be a good idea.

Parameters

mrp	target peer
block	the block

Returns: score (>= 0: feasible, negative: infeasible)

Definition at line 294 of file gnunet-service-fs_push.c.

{
  unsigned int i;
  struct GSF_PeerPerformanceData *ppd;
  struct GNUNET_PeerIdentity id;
  struct GNUNET_HashCode hc;
  uint32_t dist;
 
  ppd = GSF_get_peer_performance_data_ (mrp->peer);
  for (i = 0; i < MIGRATION_LIST_SIZE; i++)
    if (block->target_list[i] == ppd->pid)
      return -1;
  GNUNET_assert (0 != ppd->pid);
  GNUNET_PEER_resolve (ppd->pid,
                       &id);
  GNUNET_CRYPTO_hash (&id,
                      sizeof(struct GNUNET_PeerIdentity),
                      &hc);
  dist = GNUNET_CRYPTO_hash_distance_u32 (&block->query,
                                          &hc);
  /* closer distance, higher score: */
  return UINT32_MAX - dist;
}

References GNUNET_assert, GNUNET_CRYPTO_hash(), GNUNET_CRYPTO_hash_distance_u32(), GNUNET_PEER_resolve(), GSF_get_peer_performance_data_(), id, MIGRATION_LIST_SIZE, MigrationReadyPeer::peer, GSF_PeerPerformanceData::pid, MigrationReadyBlock::query, and MigrationReadyBlock::target_list.

Referenced by find_content().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ consider_gathering()

static void consider_gathering ( void )

static

If the migration task is not currently running, consider (re)scheduling it with the appropriate delay.

Definition at line 402 of file gnunet-service-fs_push.c.

{
  struct GNUNET_TIME_Relative delay;
 
  if (NULL == GSF_dsh)
    return;
  if (NULL != mig_qe)
    return;
  if (NULL != mig_task)
    return;
  if (mig_size >= MAX_MIGRATION_QUEUE)
    return;
  delay = GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS,
                                         mig_size);
  delay = GNUNET_TIME_relative_divide (delay,
                                       MAX_MIGRATION_QUEUE);
  delay = GNUNET_TIME_relative_max (delay,
                                    min_migration_delay);
  if (GNUNET_NO == value_found)
  {
    /* wait at least 5s if the datastore is empty */
    delay = GNUNET_TIME_relative_max (delay,
                                      GNUNET_TIME_relative_multiply (
                                        GNUNET_TIME_UNIT_SECONDS,
                                        5));
  }
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Scheduling gathering task (queue size: %u)\n",
              mig_size);
  mig_task = GNUNET_SCHEDULER_add_delayed (delay,
                                           &gather_migration_blocks,
                                           NULL);
}

References gather_migration_blocks(), GNUNET_ERROR_TYPE_DEBUG, GNUNET_log, GNUNET_NO, GNUNET_SCHEDULER_add_delayed(), GNUNET_TIME_relative_divide(), GNUNET_TIME_relative_max(), GNUNET_TIME_relative_multiply(), GNUNET_TIME_UNIT_SECONDS, GSF_dsh, MAX_MIGRATION_QUEUE, mig_qe, mig_size, mig_task, min_migration_delay, and value_found.

Referenced by find_content(), gather_migration_blocks(), GSF_push_init_(), and process_migration_content().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ gather_migration_blocks()

static void gather_migration_blocks ( void * cls )

static

Task that is run periodically to obtain blocks for content migration.

Parameters

cls unused

Definition at line 539 of file gnunet-service-fs_push.c.

{
  mig_task = NULL;
  if (mig_size >= MAX_MIGRATION_QUEUE)
    return;
  if (NULL == GSF_dsh)
    return;
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Asking datastore for content for replication (queue size: %u)\n",
              mig_size);
  value_found = GNUNET_NO;
  mig_qe = GNUNET_DATASTORE_get_for_replication (GSF_dsh,
                                                 0,
                                                 UINT_MAX,
                                                 &process_migration_content,
                                                 NULL);
  if (NULL == mig_qe)
    consider_gathering ();
}

References consider_gathering(), GNUNET_DATASTORE_get_for_replication(), GNUNET_ERROR_TYPE_DEBUG, GNUNET_log, GNUNET_NO, GSF_dsh, MAX_MIGRATION_QUEUE, mig_qe, mig_size, mig_task, process_migration_content(), and value_found.

Referenced by consider_gathering().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ process_migration_content()

static void process_migration_content	(	void *	cls,
		const struct GNUNET_HashCode *	key,
		size_t	size,
		const void *	data,
		enum GNUNET_BLOCK_Type	type,
		uint32_t	priority,
		uint32_t	anonymity,
		uint32_t	replication,
		struct GNUNET_TIME_Absolute	expiration,
		uint64_t	uid
	)

static

Process content offered for migration.

Parameters

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

Definition at line 453 of file gnunet-service-fs_push.c.

{
  struct MigrationReadyBlock *mb;
  struct MigrationReadyPeer *pos;
 
  mig_qe = NULL;
  if (NULL == key)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "No content found for migration...\n");
    consider_gathering ();
    return;
  }
  value_found = GNUNET_YES;
  if (GNUNET_TIME_absolute_get_remaining (expiration).rel_value_us <
      MIN_MIGRATION_CONTENT_LIFETIME.rel_value_us)
  {
    /* content will expire soon, don't bother */
    consider_gathering ();
    return;
  }
  if (type == GNUNET_BLOCK_TYPE_FS_ONDEMAND)
  {
    if (GNUNET_OK !=
        GNUNET_FS_handle_on_demand_block (key,
                                          size,
                                          data,
                                          type,
                                          priority,
                                          anonymity,
                                          replication,
                                          expiration,
                                          uid,
                                          &process_migration_content,
                                          NULL))
      consider_gathering ();
    return;
  }
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Retrieved block `%s' of type %u for migration (queue size: %u/%u)\n",
              GNUNET_h2s (key),
              type, mig_size + 1,
              MAX_MIGRATION_QUEUE);
  mb = GNUNET_malloc (sizeof(struct MigrationReadyBlock) + size);
  mb->query = *key;
  mb->expiration = expiration;
  mb->size = size;
  mb->type = type;
  GNUNET_memcpy (&mb[1], data, size);
  GNUNET_CONTAINER_DLL_insert_after (mig_head,
                                     mig_tail,
                                     mig_tail,
                                     mb);
  mig_size++;
  for (pos = peer_head; NULL != pos; pos = pos->next)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Preparing to push best content to peer %s\n",
                GNUNET_i2s (GSF_connected_peer_get_identity2_ (pos->peer)));
    if ((NULL == pos->env) &&
        (GNUNET_YES == transmit_content (pos,
                                         mb)))
    {
      break;                  /* 'mb' was freed! */
    }
  }
  consider_gathering ();
}

References anonymity, consider_gathering(), data, MigrationReadyPeer::env, expiration, MigrationReadyBlock::expiration, GNUNET_BLOCK_TYPE_FS_ONDEMAND, GNUNET_CONTAINER_DLL_insert_after, GNUNET_ERROR_TYPE_DEBUG, GNUNET_FS_handle_on_demand_block(), GNUNET_h2s(), GNUNET_i2s(), GNUNET_log, GNUNET_malloc, GNUNET_memcpy, GNUNET_OK, GNUNET_TIME_absolute_get_remaining(), GNUNET_YES, GSF_connected_peer_get_identity2_(), key, MAX_MIGRATION_QUEUE, mig_head, mig_qe, mig_size, mig_tail, MIN_MIGRATION_CONTENT_LIFETIME, MigrationReadyPeer::next, MigrationReadyPeer::peer, peer_head, process_migration_content(), MigrationReadyBlock::query, replication, size, MigrationReadyBlock::size, transmit_content(), type, MigrationReadyBlock::type, and value_found.

Referenced by gather_migration_blocks(), and process_migration_content().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ GSF_push_start_()

void GSF_push_start_ ( struct GSF_ConnectedPeer * peer )

A peer connected to us.

A peer connected to us or we are now again allowed to push content.

Start pushing content to this peer.

Parameters

peer	handle for the peer that connected

Definition at line 567 of file gnunet-service-fs_push.c.

{
  struct MigrationReadyPeer *mrp;
 
  if (GNUNET_YES != enabled)
    return;
  for (mrp = peer_head; NULL != mrp; mrp = mrp->next)
    if (mrp->peer == peer)
      break;
  if (NULL != mrp)
  {
    /* same peer added twice, must not happen */
    GNUNET_break (0);
    return;
  }
 
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Adding peer %s to list for pushing\n",
              GNUNET_i2s (GSF_connected_peer_get_identity2_ (peer)));
 
  mrp = GNUNET_new (struct MigrationReadyPeer);
  mrp->peer = peer;
  find_content (mrp);
  GNUNET_CONTAINER_DLL_insert (peer_head,
                               peer_tail,
                               mrp);
}

References enabled, find_content(), GNUNET_break, GNUNET_CONTAINER_DLL_insert, GNUNET_ERROR_TYPE_DEBUG, GNUNET_i2s(), GNUNET_log, GNUNET_new, GNUNET_YES, GSF_connected_peer_get_identity2_(), MigrationReadyPeer::next, MigrationReadyPeer::peer, peer_head, and peer_tail.

Referenced by peer_respect_cb(), and revive_migration().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ GSF_push_stop_()

void GSF_push_stop_ ( struct GSF_ConnectedPeer * peer )

A peer disconnected from us.

A peer disconnected from us or asked us to stop pushing content for a while.

Stop pushing content to this peer.

Parameters

peer	handle for the peer that disconnected

Definition at line 603 of file gnunet-service-fs_push.c.

{
  struct MigrationReadyPeer *pos;
 
  for (pos = peer_head; NULL != pos; pos = pos->next)
    if (pos->peer == peer)
      break;
  if (NULL == pos)
    return;
  if (NULL != pos->env)
    GNUNET_MQ_send_cancel (pos->env);
  GNUNET_CONTAINER_DLL_remove (peer_head,
                               peer_tail,
                               pos);
  GNUNET_free (pos);
}

References MigrationReadyPeer::env, GNUNET_CONTAINER_DLL_remove, GNUNET_free, GNUNET_MQ_send_cancel(), MigrationReadyPeer::next, MigrationReadyPeer::peer, peer_head, and peer_tail.

Referenced by GSF_peer_disconnect_handler(), and handle_p2p_migration_stop().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ GSF_push_init_()

void GSF_push_init_ ( void )

Setup the module.

Definition at line 625 of file gnunet-service-fs_push.c.

{
  enabled =
    GNUNET_CONFIGURATION_get_value_yesno (GSF_cfg,
                                          "FS",
                                          "CONTENT_PUSHING");
  if (GNUNET_YES != enabled)
    return;
 
  if (GNUNET_OK !=
      GNUNET_CONFIGURATION_get_value_time (GSF_cfg,
                                           "fs",
                                           "MIN_MIGRATION_DELAY",
                                           &min_migration_delay))
  {
    GNUNET_log_config_invalid (GNUNET_ERROR_TYPE_WARNING,
                               "fs",
                               "MIN_MIGRATION_DELAY",
                               _ ("time required, content pushing disabled"));
    return;
  }
  consider_gathering ();
}

References _, consider_gathering(), enabled, GNUNET_CONFIGURATION_get_value_time(), GNUNET_CONFIGURATION_get_value_yesno(), GNUNET_ERROR_TYPE_WARNING, GNUNET_log_config_invalid(), GNUNET_OK, GNUNET_YES, GSF_cfg, and min_migration_delay.

Referenced by run().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ GSF_push_done_()

void GSF_push_done_ ( void )

Shutdown the module.

Definition at line 654 of file gnunet-service-fs_push.c.

{
  if (NULL != mig_task)
  {
    GNUNET_SCHEDULER_cancel (mig_task);
    mig_task = NULL;
  }
  if (NULL != mig_qe)
  {
    GNUNET_DATASTORE_cancel (mig_qe);
    mig_qe = NULL;
  }
  while (NULL != mig_head)
    delete_migration_block (mig_head);
  GNUNET_assert (0 == mig_size);
}