/*
This file is part of GNUnet.
- (C) 2011 Christian Grothoff (and other contributing authors)
+ Copyright (C) 2011 Christian Grothoff (and other contributing authors)
GNUnet is free software; you can redistribute it and/or modify
it under the terms of the GNU General Public License as published
You should have received a copy of the GNU General Public License
along with GNUnet; see the file COPYING. If not, write to the
- Free Software Foundation, Inc., 59 Temple Place - Suite 330,
- Boston, MA 02111-1307, USA.
+ Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor,
+ Boston, MA 02110-1301, USA.
*/
/**
#include "gnunet-service-fs_push.h"
-#define DEBUG_FS_MIGRATION GNUNET_EXTRA_LOGGING
+/**
+ * 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?
/**
* Query for the block.
*/
- GNUNET_HashCode query;
+ struct GNUNET_HashCode query;
/**
* When does this block expire?
/**
* ID of task that collects blocks for migration.
*/
-static GNUNET_SCHEDULER_TaskIdentifier mig_task;
+static struct GNUNET_SCHEDULER_Task * mig_task;
/**
* What is the maximum frequency at which we are allowed to
*/
static int enabled;
+/**
+ * Did we find anything in the datastore?
+ */
+static int value_found;
+
/**
* Delete the given migration block.
/**
- * Transmit the message currently scheduled for
- * transmission.
+ * Transmit the message currently scheduled for transmission.
*
- * @param cls the 'struct MigrationReadyPeer'
- * @param buf_size number of bytes available in buf
+ * @param cls the `struct MigrationReadyPeer`
+ * @param buf_size number of bytes available in @a buf
* @param buf where to copy the message, NULL on error (peer disconnect)
- * @return number of bytes copied to 'buf', can be 0 (without indicating an error)
+ * @return number of bytes copied to @a 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;
peer->th = NULL;
msg = peer->msg;
peer->msg = NULL;
- if (buf == NULL)
+ if (NULL == buf)
{
-#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_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 %s\n",
+ msize,
+ GNUNET_i2s (GSF_connected_peer_get_identity2_(peer->peer)));
find_content (peer);
return msize;
}
*
* @param peer target peer
* @param block the block
- * @return GNUNET_YES if the block was deleted (!)
+ * @return #GNUNET_YES if the block was deleted (!)
*/
static int
transmit_content (struct MigrationReadyPeer *peer,
{
ret = GNUNET_NO;
}
-#if DEBUG_FS_MIGRATION
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,
+ "Asking for transmission of %u bytes to %s for migration\n",
+ msize,
+ GNUNET_i2s (GSF_connected_peer_get_identity2_(peer->peer)));
+ 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;
struct GSF_PeerPerformanceData *ppd;
struct GNUNET_PeerIdentity id;
+ struct GNUNET_HashCode hc;
uint32_t dist;
ppd = GSF_get_peer_performance_data_ (peer->peer);
return -1;
GNUNET_assert (0 != ppd->pid);
GNUNET_PEER_resolve (ppd->pid, &id);
- dist = GNUNET_CRYPTO_hash_distance_u32 (&block->query, &id.hashPubKey);
+ 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;
}
{
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 */
consider_gathering ();
return;
}
-#if DEBUG_FS_MIGRATION
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Preparing to push best content to peer %s\n");
-#endif
+ "Preparing to push best content to peer\n");
transmit_content (mrp, best);
}
{
struct GNUNET_TIME_Relative delay;
- if (GSF_dsh == NULL)
+ if (NULL == GSF_dsh)
return;
- if (mig_qe != NULL)
+ if (NULL != mig_qe)
return;
- if (mig_task != GNUNET_SCHEDULER_NO_TASK)
+ 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 DEBUG_FS_MIGRATION
+ 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);
-#endif
+ "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 struct 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 (NULL == key)
{
-#if DEBUG_FS_MIGRATION
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "No content found for migration...\n");
-#endif
+ 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)
+ 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 ();
if (type == GNUNET_BLOCK_TYPE_FS_ONDEMAND)
{
if (GNUNET_OK !=
- GNUNET_FS_handle_on_demand_block (key, size, data, type, priority,
- anonymity, expiration, uid,
+ 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
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
+ 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)
+ for (pos = peer_head; NULL != pos; pos = pos->next)
{
if (NULL == pos->th)
{
-#if DEBUG_FS_MIGRATION
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Preparing to push best content to peer\n");
-#endif
+ "Preparing to push best content to peer %s\n",
+ GNUNET_i2s (GSF_connected_peer_get_identity2_(pos->peer)));
if (GNUNET_YES == transmit_content (pos, mb))
break; /* 'mb' was freed! */
}
- pos = pos->next;
}
consider_gathering ();
}
gather_migration_blocks (void *cls,
const struct GNUNET_SCHEDULER_TaskContext *tc)
{
- mig_task = GNUNET_SCHEDULER_NO_TASK;
+ mig_task = NULL;
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 ();
- }
+ 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,
+ GNUNET_TIME_UNIT_FOREVER_REL,
+ &process_migration_content, NULL);
+ if (NULL == mig_qe)
+ consider_gathering ();
}
if (GNUNET_YES != enabled)
return;
- mrp = GNUNET_malloc (sizeof (struct MigrationReadyPeer));
+ 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);
+ GNUNET_CONTAINER_DLL_insert (peer_head,
+ peer_tail,
+ mrp);
}
{
struct MigrationReadyPeer *pos;
- pos = peer_head;
- while (pos != NULL)
- {
+ for (pos = peer_head; NULL != pos; pos = pos->next)
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;
+ break;
+ if (NULL == pos)
+ return;
+ 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);
}
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");
+ GNUNET_log_config_invalid (GNUNET_ERROR_TYPE_WARNING,
+ "fs", "MIN_MIGRATION_DELAY",
+ _("time required, content pushing disabled"));
return;
}
consider_gathering ();
void
GSF_push_done_ ()
{
- if (GNUNET_SCHEDULER_NO_TASK != mig_task)
+ if (NULL != mig_task)
{
GNUNET_SCHEDULER_cancel (mig_task);
- mig_task = GNUNET_SCHEDULER_NO_TASK;
+ mig_task = NULL;
}
if (NULL != mig_qe)
{
projects / oweals / gnunet.git / blobdiff