#include "gnunet-service-fs_push.h"
-#define DEBUG_FS_MIGRATION GNUNET_NO
+/**
+ * Maximum number of blocks we keep in memory for migration.
+ */
+#define MAX_MIGRATION_QUEUE 8
+
+/**
+ * Blocks are at most migrated to this number of peers
+ * plus one, each time they are fetched from the database.
+ */
+#define MIGRATION_LIST_SIZE 2
/**
- * How long must content remain valid for us to consider it for migration?
+ * 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
GNUNET_HashCode query;
/**
- * When does this block expire?
+ * When does this block expire?
*/
struct GNUNET_TIME_Absolute expiration;
* Handle to peer.
*/
struct GSF_ConnectedPeer *peer;
-
+
/**
* Handle for current transmission request,
* or NULL for none.
static void
delete_migration_block (struct MigrationReadyBlock *mb)
{
- GNUNET_CONTAINER_DLL_remove (mig_head,
- mig_tail,
- mb);
- GNUNET_PEER_decrement_rcs (mb->target_list,
- MIGRATION_LIST_SIZE);
+ GNUNET_CONTAINER_DLL_remove (mig_head, mig_tail, mb);
+ GNUNET_PEER_decrement_rcs (mb->target_list, MIGRATION_LIST_SIZE);
mig_size--;
GNUNET_free (mb);
}
/**
* Find content for migration to this peer.
- */
+ */
static void
find_content (struct MigrationReadyPeer *mrp);
* @return number of bytes copied to 'buf', can be 0 (without indicating an error)
*/
static size_t
-transmit_message (void *cls,
- size_t buf_size,
- void *buf)
+transmit_message (void *cls, size_t buf_size, void *buf)
{
struct MigrationReadyPeer *peer = cls;
struct PutMessage *msg;
msg = peer->msg;
peer->msg = NULL;
if (buf == NULL)
- {
-#if DEBUG_FS_MIGRATION
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Failed to migrate content to another peer (disconnect)\n");
-#endif
- GNUNET_free (msg);
- return 0;
- }
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Failed to migrate content to another peer (disconnect)\n");
+ GNUNET_free (msg);
+ return 0;
+ }
msize = ntohs (msg->header.size);
GNUNET_assert (msize <= buf_size);
memcpy (buf, msg, msize);
GNUNET_free (msg);
-#if DEBUG_FS_MIGRATION
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Pushing %u bytes to another peer\n",
- msize);
-#endif
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Pushing %u bytes to another peer\n",
+ msize);
find_content (peer);
return msize;
}
*/
static int
transmit_content (struct MigrationReadyPeer *peer,
- struct MigrationReadyBlock *block)
+ struct MigrationReadyBlock *block)
{
size_t msize;
struct PutMessage *msg;
- unsigned int i;
+ unsigned int i;
struct GSF_PeerPerformanceData *ppd;
int ret;
msg->header.size = htons (msize);
msg->type = htonl (block->type);
msg->expiration = GNUNET_TIME_absolute_hton (block->expiration);
- memcpy (&msg[1],
- &block[1],
- block->size);
+ memcpy (&msg[1], &block[1], block->size);
peer->msg = msg;
- for (i=0;i<MIGRATION_LIST_SIZE;i++)
+ for (i = 0; i < MIGRATION_LIST_SIZE; i++)
+ {
+ if (block->target_list[i] == 0)
{
- if (block->target_list[i] == 0)
- {
- block->target_list[i] = ppd->pid;
- GNUNET_PEER_change_rc (block->target_list[i], 1);
- break;
- }
+ 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;
- }
+ {
+ delete_migration_block (block);
+ ret = GNUNET_YES;
+ }
else
- {
- ret = GNUNET_NO;
- }
-#if DEBUG_FS_MIGRATION
+ {
+ ret = GNUNET_NO;
+ }
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Asking for transmission of %u bytes for migration\n",
- msize);
-#endif
- peer->th = GSF_peer_transmit_ (peer->peer,
- GNUNET_NO,
- 0 /* priority */,
- GNUNET_TIME_UNIT_FOREVER_REL,
- msize,
- &transmit_message,
- peer);
+ "Asking for transmission of %u bytes for migration\n", msize);
+ peer->th = GSF_peer_transmit_ (peer->peer, GNUNET_NO, 0 /* priority */ ,
+ GNUNET_TIME_UNIT_FOREVER_REL, msize,
+ &transmit_message, peer);
return ret;
}
{
unsigned int i;
- for (i=0;i<MIGRATION_LIST_SIZE;i++)
+ for (i = 0; i < MIGRATION_LIST_SIZE; i++)
if (block->target_list[i] == 0)
return i;
return i;
/**
* Check if sending this block to this peer would
- * be a good idea.
+ * be a good idea.
*
* @param peer target peer
* @param block the block
*/
static long
score_content (struct MigrationReadyPeer *peer,
- struct MigrationReadyBlock *block)
+ struct MigrationReadyBlock *block)
{
unsigned int i;
struct GSF_PeerPerformanceData *ppd;
uint32_t dist;
ppd = GSF_get_peer_performance_data_ (peer->peer);
- for (i=0;i<MIGRATION_LIST_SIZE;i++)
+ for (i = 0; i < MIGRATION_LIST_SIZE; i++)
if (block->target_list[i] == ppd->pid)
return -1;
- GNUNET_PEER_resolve (ppd->pid,
- &id);
- dist = GNUNET_CRYPTO_hash_distance_u32 (&block->query,
- &id.hashPubKey);
+ GNUNET_assert (0 != ppd->pid);
+ GNUNET_PEER_resolve (ppd->pid, &id);
+ dist = GNUNET_CRYPTO_hash_distance_u32 (&block->query, &id.hashPubKey);
/* closer distance, higher score: */
return UINT32_MAX - dist;
}
* Find content for migration to this peer.
*
* @param mrp peer to find content for
- */
+ */
static void
find_content (struct MigrationReadyPeer *mrp)
{
best_score = -1;
pos = mig_head;
while (NULL != pos)
+ {
+ score = score_content (mrp, pos);
+ if (score > best_score)
{
- score = score_content (mrp, pos);
- if (score > best_score)
- {
- best_score = score;
- best = pos;
- }
- pos = pos->next;
+ best_score = score;
+ best = pos;
}
- if (NULL == best)
+ pos = pos->next;
+ }
+ if (NULL == best)
+ {
+ if (mig_size < MAX_MIGRATION_QUEUE)
{
- if (mig_size < MAX_MIGRATION_QUEUE)
- {
-#if DEBUG_FS_MIGRATION
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "No content found for pushing, waiting for queue to fill\n");
-#endif
- return; /* will fill up eventually... */
- }
-#if DEBUG_FS_MIGRATION
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "No suitable content found, purging content from full queue\n");
-#endif
- /* 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;
+ "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;
}
-#if DEBUG_FS_MIGRATION
+ 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");
-#endif
+ "Preparing to push best content to peer\n");
transmit_content (mrp, best);
}
/**
* Task that is run periodically to obtain blocks for content
* migration
- *
+ *
* @param cls unused
* @param tc scheduler context (also unused)
*/
static void
gather_migration_blocks (void *cls,
- const struct GNUNET_SCHEDULER_TaskContext *tc);
+ const struct GNUNET_SCHEDULER_TaskContext *tc);
/**
return;
if (mig_task != GNUNET_SCHEDULER_NO_TASK)
return;
- if (mig_size >= MAX_MIGRATION_QUEUE)
+ 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 DEBUG_FS_MIGRATION
+ 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);
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Scheduling gathering task (queue size: %u)\n",
- mig_size);
-#endif
- mig_task = GNUNET_SCHEDULER_add_delayed (delay,
- &gather_migration_blocks,
- NULL);
+ "Scheduling gathering task (queue size: %u)\n", mig_size);
+ mig_task =
+ GNUNET_SCHEDULER_add_delayed (delay, &gather_migration_blocks, NULL);
}
* maybe 0 if no unique identifier is available
*/
static void
-process_migration_content (void *cls,
- const GNUNET_HashCode *key,
- size_t size,
- const void *data,
- enum GNUNET_BLOCK_Type type,
- uint32_t priority,
- uint32_t anonymity,
- struct GNUNET_TIME_Absolute
- expiration, uint64_t uid)
+process_migration_content (void *cls, const GNUNET_HashCode * key, size_t size,
+ const void *data, enum GNUNET_BLOCK_Type type,
+ uint32_t priority, uint32_t anonymity,
+ struct GNUNET_TIME_Absolute expiration, uint64_t uid)
{
struct MigrationReadyBlock *mb;
struct MigrationReadyPeer *pos;
-
+
mig_qe = NULL;
if (key == NULL)
- {
-#if DEBUG_FS_MIGRATION
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "No content found for migration...\n");
-#endif
- consider_gathering ();
- return;
- }
- if (GNUNET_TIME_absolute_get_remaining (expiration).rel_value <
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "No content found for migration...\n");
+ consider_gathering ();
+ return;
+ }
+ if (GNUNET_TIME_absolute_get_remaining (expiration).rel_value <
MIN_MIGRATION_CONTENT_LIFETIME.rel_value)
- {
- /* content will expire soon, don't bother */
- consider_gathering ();
- return;
- }
+ {
+ /* 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,
- expiration, uid,
- &process_migration_content,
- NULL))
- consider_gathering ();
- return;
- }
-#if DEBUG_FS_MIGRATION
+ {
+ if (GNUNET_OK !=
+ GNUNET_FS_handle_on_demand_block (key, size, data, type, priority,
+ anonymity, 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);
-#endif
+ "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;
memcpy (&mb[1], data, size);
- GNUNET_CONTAINER_DLL_insert_after (mig_head,
- mig_tail,
- mig_tail,
- mb);
+ GNUNET_CONTAINER_DLL_insert_after (mig_head, mig_tail, mig_tail, mb);
mig_size++;
pos = peer_head;
while (pos != NULL)
+ {
+ if (NULL == pos->th)
{
- if (NULL == pos->th)
- {
-#if DEBUG_FS_MIGRATION
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Preparing to push best content to peer\n");
-#endif
- if (GNUNET_YES == transmit_content (pos, mb))
- break; /* 'mb' was freed! */
- }
- pos = pos->next;
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Preparing to push best content to peer\n");
+ if (GNUNET_YES == transmit_content (pos, mb))
+ break; /* 'mb' was freed! */
}
+ pos = pos->next;
+ }
consider_gathering ();
}
/**
* Task that is run periodically to obtain blocks for content
* migration
- *
+ *
* @param cls unused
* @param tc scheduler context (also unused)
*/
static void
gather_migration_blocks (void *cls,
- const struct GNUNET_SCHEDULER_TaskContext *tc)
+ const struct GNUNET_SCHEDULER_TaskContext *tc)
{
mig_task = GNUNET_SCHEDULER_NO_TASK;
- if (mig_size >= MAX_MIGRATION_QUEUE)
+ if (mig_size >= MAX_MIGRATION_QUEUE)
return;
if (GSF_dsh != NULL)
- {
-#if DEBUG_FS_MIGRATION
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Asking datastore for content for replication (queue size: %u)\n",
- mig_size);
-#endif
- mig_qe = GNUNET_DATASTORE_get_for_replication (GSF_dsh,
- 0, UINT_MAX,
- GNUNET_TIME_UNIT_FOREVER_REL,
- &process_migration_content, NULL);
- if (NULL == mig_qe)
- consider_gathering ();
- }
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Asking datastore for content for replication (queue size: %u)\n",
+ mig_size);
+ mig_qe =
+ GNUNET_DATASTORE_get_for_replication (GSF_dsh, 0, UINT_MAX,
+ GNUNET_TIME_UNIT_FOREVER_REL,
+ &process_migration_content, NULL);
+ if (NULL == mig_qe)
+ consider_gathering ();
+ }
}
mrp = GNUNET_malloc (sizeof (struct MigrationReadyPeer));
mrp->peer = peer;
find_content (mrp);
- GNUNET_CONTAINER_DLL_insert (peer_head,
- peer_tail,
- mrp);
+ GNUNET_CONTAINER_DLL_insert (peer_head, peer_tail, mrp);
}
pos = peer_head;
while (pos != NULL)
+ {
+ if (pos->peer == peer)
{
- if (pos->peer == peer)
- {
- GNUNET_CONTAINER_DLL_remove (peer_head,
- peer_tail,
- pos);
- if (NULL != pos->th)
- {
- GSF_peer_transmit_cancel_ (pos->th);
- pos->th = NULL;
- }
- if (NULL != pos->msg)
- {
- GNUNET_free (pos->msg);
- pos->msg = NULL;
- }
- GNUNET_free (pos);
- return;
- }
- pos = pos->next;
+ GNUNET_CONTAINER_DLL_remove (peer_head, peer_tail, pos);
+ if (NULL != pos->th)
+ {
+ GSF_peer_transmit_cancel_ (pos->th);
+ pos->th = NULL;
+ }
+ if (NULL != pos->msg)
+ {
+ GNUNET_free (pos->msg);
+ pos->msg = NULL;
+ }
+ GNUNET_free (pos);
+ return;
}
+ pos = pos->next;
+ }
}
void
GSF_push_init_ ()
{
- enabled = GNUNET_CONFIGURATION_get_value_yesno (GSF_cfg,
- "FS",
- "CONTENT_PUSHING");
+ 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 (GNUNET_ERROR_TYPE_WARNING,
- _("Invalid value specified for option `%s' in section `%s', content pushing disabled\n"),
- "MIN_MIGRATION_DELAY",
- "fs");
- return;
- }
+
+ if (GNUNET_OK !=
+ GNUNET_CONFIGURATION_get_value_time (GSF_cfg, "fs", "MIN_MIGRATION_DELAY",
+ &min_migration_delay))
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+ _
+ ("Invalid value specified for option `%s' in section `%s', content pushing disabled\n"),
+ "MIN_MIGRATION_DELAY", "fs");
+ return;
+ }
consider_gathering ();
}
GSF_push_done_ ()
{
if (GNUNET_SCHEDULER_NO_TASK != mig_task)
- {
- GNUNET_SCHEDULER_cancel (mig_task);
- mig_task = GNUNET_SCHEDULER_NO_TASK;
- }
+ {
+ GNUNET_SCHEDULER_cancel (mig_task);
+ mig_task = GNUNET_SCHEDULER_NO_TASK;
+ }
if (NULL != mig_qe)
- {
- GNUNET_DATASTORE_cancel (mig_qe);
- mig_qe = NULL;
- }
+ {
+ GNUNET_DATASTORE_cancel (mig_qe);
+ mig_qe = NULL;
+ }
while (NULL != mig_head)
delete_migration_block (mig_head);
GNUNET_assert (0 == mig_size);
projects / oweals / gnunet.git / blobdiff