GNUnet is free software; you can redistribute it and/or modify
it under the terms of the GNU General Public License as published
- by the Free Software Foundation; either version 2, or (at your
+ by the Free Software Foundation; either version 3, or (at your
option) any later version.
GNUnet is distributed in the hope that it will be useful, but
* @author Christian Grothoff
*
* TODO:
- * - trust not properly received and pushed back to peerinfo!
- * - bound_priority by priorities used by other peers
- * - have a way to drop queries based on load
+ * - track per-peer request latency (using new load API)
+ * - consider more precise latency estimation (per-peer & request) -- again load API?
+ * - implement test_load_too_high, make decision priority-based, implement forwarding, etc.
* - introduce random latency in processing
- * - consider more precise latency estimation (per-peer & request)
- * - better algorithm for priority selection for requests we initiate?
- * - tell other peers to stop migration if our PUTs fail (or if
- * we don't support migration per configuration?)
* - more statistics
*/
#include "platform.h"
#include <float.h>
#include "gnunet_constants.h"
#include "gnunet_core_service.h"
+#include "gnunet_dht_service.h"
#include "gnunet_datastore_service.h"
+#include "gnunet_load_lib.h"
#include "gnunet_peer_lib.h"
#include "gnunet_protocols.h"
#include "gnunet_signatures.h"
*/
#define MAX_QUEUE_PER_PEER 16
+/**
+ * Size for the hash map for DHT requests from the FS
+ * service. Should be about the number of concurrent
+ * DHT requests we plan to make.
+ */
+#define FS_DHT_HT_SIZE 1024
+
+/**
+ * How often do we flush trust values to disk?
+ */
+#define TRUST_FLUSH_FREQ GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MINUTES, 5)
+
/**
* Inverse of the probability that we will submit the same query
* to the same peer again. If the same peer already got the query
*/
#define MAX_TRANSMIT_DELAY GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 45)
-
-
/**
* Maximum number of requests (from other peers) that we're
* willing to have pending at any given point in time.
- * FIXME: set from configuration.
*/
-static uint64_t max_pending_requests = (32 * 1024);
+static unsigned long long max_pending_requests = (32 * 1024);
/**
*/
struct PendingMessage;
-/**
- * Our connection to the datastore.
- */
-static struct GNUNET_DATASTORE_Handle *dsh;
-
-
/**
* Function called upon completion of a transmission.
*
*/
struct GNUNET_TIME_Relative avg_delay;
+ /**
+ * Point in time until which this peer does not want us to migrate content
+ * to it.
+ */
+ struct GNUNET_TIME_Absolute migration_blocked;
+
+ /**
+ * Time until when we blocked this peer from migrating
+ * data to us.
+ */
+ struct GNUNET_TIME_Absolute last_migration_block;
+
/**
* Handle for an active request for transmission to this
* peer, or NULL.
uint64_t inc_preference;
/**
- * Trust delta to still commit to the system.
+ * Trust rating for this peer
+ */
+ uint32_t trust;
+
+ /**
+ * Trust rating for this peer on disk.
*/
- uint32_t trust_delta;
+ uint32_t disk_trust;
/**
* The peer's identity.
};
+/**
+ * Our connection to the datastore.
+ */
+static struct GNUNET_DATASTORE_Handle *dsh;
+
+/**
+ * Our block context.
+ */
+static struct GNUNET_BLOCK_Context *block_ctx;
+
+/**
+ * Our block configuration.
+ */
+static struct GNUNET_CONFIGURATION_Handle *block_cfg;
+
/**
* Our scheduler.
*/
*/
static struct GNUNET_DATASTORE_QueueEntry *mig_qe;
+/**
+ * Where do we store trust information?
+ */
+static char *trustDirectory;
+
/**
* ID of task that collects blocks for migration.
*/
*/
static struct GNUNET_TIME_Relative min_migration_delay;
+/**
+ * Handle for DHT operations.
+ */
+static struct GNUNET_DHT_Handle *dht_handle;
+
/**
* Size of the doubly-linked list of migration blocks.
*/
*/
static int active_migration;
+/**
+ * Typical priorities we're seeing from other peers right now. Since
+ * most priorities will be zero, this value is the weighted average of
+ * non-zero priorities seen "recently". In order to ensure that new
+ * values do not dramatically change the ratio, values are first
+ * "capped" to a reasonable range (+N of the current value) and then
+ * averaged into the existing value by a ratio of 1:N. Hence
+ * receiving the largest possible priority can still only raise our
+ * "current_priorities" by at most 1.
+ */
+static double current_priorities;
+
+/**
+ * Datastore 'GET' load tracking.
+ */
+static struct GNUNET_LOAD_Value *datastore_get_load;
+
+/**
+ * Datastore 'PUT' load tracking.
+ */
+static struct GNUNET_LOAD_Value *datastore_put_load;
+
+
+/**
+ * We've just now completed a datastore request. Update our
+ * datastore load calculations.
+ *
+ * @param start time when the datastore request was issued
+ */
+static void
+update_datastore_delays (struct GNUNET_TIME_Absolute start)
+{
+ struct GNUNET_TIME_Relative delay;
+
+ delay = GNUNET_TIME_absolute_get_duration (start);
+ GNUNET_LOAD_update (datastore_get_load,
+ delay.value);
+}
+
+
+/**
+ * Get the filename under which we would store the GNUNET_HELLO_Message
+ * for the given host and protocol.
+ * @return filename of the form DIRECTORY/HOSTID
+ */
+static char *
+get_trust_filename (const struct GNUNET_PeerIdentity *id)
+{
+ struct GNUNET_CRYPTO_HashAsciiEncoded fil;
+ char *fn;
+
+ GNUNET_CRYPTO_hash_to_enc (&id->hashPubKey, &fil);
+ GNUNET_asprintf (&fn, "%s%s%s", trustDirectory, DIR_SEPARATOR_STR, &fil);
+ return fn;
+}
+
+
/**
* Transmit messages by copying it to the target buffer
/* ******************* clean up functions ************************ */
-
/**
* Delete the given migration block.
*
unsigned int repl;
/* consider 'cp' as a migration target for mb */
+ if (GNUNET_TIME_absolute_get_remaining (cp->migration_blocked).value > 0)
+ return GNUNET_YES; /* peer has requested no migration! */
if (mb != NULL)
{
GNUNET_PEER_resolve (cp->pid,
{
struct GNUNET_TIME_Relative delay;
+ if (dsh == NULL)
+ return;
if (mig_qe != NULL)
return;
if (mig_task != GNUNET_SCHEDULER_NO_TASK)
consider_migration_gathering ();
return;
}
- if (type == GNUNET_BLOCK_TYPE_ONDEMAND)
+ if (type == GNUNET_BLOCK_TYPE_FS_ONDEMAND)
{
if (GNUNET_OK !=
GNUNET_FS_handle_on_demand_block (key, size, data,
expiration, uid,
&process_migration_content,
NULL))
- GNUNET_DATASTORE_get_next (dsh, GNUNET_YES);
+ {
+ GNUNET_DATASTORE_get_next (dsh, GNUNET_YES);
+ }
return;
}
#if DEBUG_FS
const struct GNUNET_SCHEDULER_TaskContext *tc)
{
mig_task = GNUNET_SCHEDULER_NO_TASK;
- mig_qe = GNUNET_DATASTORE_get_random (dsh, 0, -1,
- GNUNET_TIME_UNIT_FOREVER_REL,
- &process_migration_content, NULL);
- GNUNET_assert (mig_qe != NULL);
+ if (dsh != NULL)
+ {
+ mig_qe = GNUNET_DATASTORE_get_random (dsh, 0, -1,
+ GNUNET_TIME_UNIT_FOREVER_REL,
+ &process_migration_content, NULL);
+ GNUNET_assert (mig_qe != NULL);
+ }
}
TransmissionContinuation cont;
void *cont_cls;
- GNUNET_assert (pml->pm == pm);
- GNUNET_assert ( (tpid == 0) || (tpid == pml->target->pid) );
- cont = pm->cont;
- cont_cls = pm->cont_cls;
- destroy_pending_message_list_entry (pml);
- cont (cont_cls, tpid);
+ if (pml != NULL)
+ {
+ GNUNET_assert (pml->pm == pm);
+ GNUNET_assert ( (tpid == 0) || (tpid == pml->target->pid) );
+ cont = pm->cont;
+ cont_cls = pm->cont_cls;
+ destroy_pending_message_list_entry (pml);
+ }
+ else
+ {
+ GNUNET_free (pm);
+ }
+ if (cont != NULL)
+ cont (cont_cls, tpid);
}
{
struct ConnectedPeer *cp;
struct MigrationReadyBlock *pos;
+ char *fn;
+ uint32_t trust;
cp = GNUNET_malloc (sizeof (struct ConnectedPeer));
cp->pid = GNUNET_PEER_intern (peer);
+
+ fn = get_trust_filename (peer);
+ if ((GNUNET_DISK_file_test (fn) == GNUNET_YES) &&
+ (sizeof (trust) == GNUNET_DISK_fn_read (fn, &trust, sizeof (trust))))
+ cp->disk_trust = cp->trust = ntohl (trust);
+ GNUNET_free (fn);
+
GNUNET_break (GNUNET_OK ==
GNUNET_CONTAINER_multihashmap_put (connected_peers,
&peer->hashPubKey,
}
+/**
+ * Increase the host credit by a value.
+ *
+ * @param host which peer to change the trust value on
+ * @param value is the int value by which the
+ * host credit is to be increased or decreased
+ * @returns the actual change in trust (positive or negative)
+ */
+static int
+change_host_trust (struct ConnectedPeer *host, int value)
+{
+ unsigned int old_trust;
+
+ if (value == 0)
+ return 0;
+ GNUNET_assert (host != NULL);
+ old_trust = host->trust;
+ if (value > 0)
+ {
+ if (host->trust + value < host->trust)
+ {
+ value = UINT32_MAX - host->trust;
+ host->trust = UINT32_MAX;
+ }
+ else
+ host->trust += value;
+ }
+ else
+ {
+ if (host->trust < -value)
+ {
+ value = -host->trust;
+ host->trust = 0;
+ }
+ else
+ host->trust += value;
+ }
+ return value;
+}
+
+
+/**
+ * Write host-trust information to a file - flush the buffer entry!
+ */
+static int
+flush_trust (void *cls,
+ const GNUNET_HashCode *key,
+ void *value)
+{
+ struct ConnectedPeer *host = value;
+ char *fn;
+ uint32_t trust;
+ struct GNUNET_PeerIdentity pid;
+
+ if (host->trust == host->disk_trust)
+ return GNUNET_OK; /* unchanged */
+ GNUNET_PEER_resolve (host->pid,
+ &pid);
+ fn = get_trust_filename (&pid);
+ if (host->trust == 0)
+ {
+ if ((0 != UNLINK (fn)) && (errno != ENOENT))
+ GNUNET_log_strerror_file (GNUNET_ERROR_TYPE_WARNING |
+ GNUNET_ERROR_TYPE_BULK, "unlink", fn);
+ }
+ else
+ {
+ trust = htonl (host->trust);
+ if (sizeof(uint32_t) == GNUNET_DISK_fn_write (fn, &trust,
+ sizeof(uint32_t),
+ GNUNET_DISK_PERM_USER_READ | GNUNET_DISK_PERM_USER_WRITE
+ | GNUNET_DISK_PERM_GROUP_READ | GNUNET_DISK_PERM_OTHER_READ))
+ host->disk_trust = host->trust;
+ }
+ GNUNET_free (fn);
+ return GNUNET_OK;
+}
+
+/**
+ * Call this method periodically to scan data/hosts for new hosts.
+ */
+static void
+cron_flush_trust (void *cls,
+ const struct GNUNET_SCHEDULER_TaskContext *tc)
+{
+
+ if (NULL == connected_peers)
+ return;
+ GNUNET_CONTAINER_multihashmap_iterate (connected_peers,
+ &flush_trust,
+ NULL);
+ if (NULL == tc)
+ return;
+ if (0 != (tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN))
+ return;
+ GNUNET_SCHEDULER_add_delayed (tc->sched,
+ TRUST_FLUSH_FREQ, &cron_flush_trust, NULL);
+}
+
/**
* Free (each) request made by the peer.
&consider_migration,
pos);
}
- if (cp->trust_delta > 0)
- {
- /* FIXME: push trust back to peerinfo!
- (need better peerinfo API!) */
- }
GNUNET_PEER_change_rc (cp->pid, -1);
GNUNET_PEER_decrement_rcs (cp->last_p2p_replies, P2P_SUCCESS_LIST_SIZE);
if (NULL != cp->cth)
while (client_list != NULL)
handle_client_disconnect (NULL,
client_list->client);
+ cron_flush_trust (NULL, NULL);
GNUNET_CONTAINER_multihashmap_iterate (connected_peers,
&clean_peer,
NULL);
GNUNET_STATISTICS_destroy (stats, GNUNET_NO);
stats = NULL;
}
- GNUNET_DATASTORE_disconnect (dsh,
- GNUNET_NO);
+ if (dsh != NULL)
+ {
+ GNUNET_DATASTORE_disconnect (dsh,
+ GNUNET_NO);
+ dsh = NULL;
+ }
while (mig_head != NULL)
delete_migration_block (mig_head);
GNUNET_assert (0 == mig_size);
- dsh = NULL;
+ GNUNET_DHT_disconnect (dht_handle);
+ dht_handle = NULL;
+ GNUNET_LOAD_value_free (datastore_get_load);
+ datastore_get_load = NULL;
+ GNUNET_LOAD_value_free (datastore_put_load);
+ datastore_put_load = NULL;
+ GNUNET_BLOCK_context_destroy (block_ctx);
+ block_ctx = NULL;
+ GNUNET_CONFIGURATION_destroy (block_cfg);
+ block_cfg = NULL;
sched = NULL;
cfg = NULL;
+ GNUNET_free_non_null (trustDirectory);
+ trustDirectory = NULL;
}
GNUNET_PEER_change_rc (mb->target_list[i], -1);
mb->target_list[i] = 0;
mb->used_targets++;
+ memset (&migm, 0, sizeof (migm));
migm.header.size = htons (sizeof (migm) + mb->size);
migm.header.type = htons (GNUNET_MESSAGE_TYPE_FS_PUT);
migm.type = htonl (mb->type);
GNUNET_assert (pm->next == NULL);
GNUNET_assert (pm->pml == NULL);
- pml = GNUNET_malloc (sizeof (struct PendingMessageList));
- pml->req = pr;
- pml->target = cp;
- pml->pm = pm;
- pm->pml = pml;
- GNUNET_CONTAINER_DLL_insert (pr->pending_head,
- pr->pending_tail,
- pml);
+ if (pr != NULL)
+ {
+ pml = GNUNET_malloc (sizeof (struct PendingMessageList));
+ pml->req = pr;
+ pml->target = cp;
+ pml->pm = pm;
+ pm->pml = pml;
+ GNUNET_CONTAINER_DLL_insert (pr->pending_head,
+ pr->pending_tail,
+ pml);
+ }
pos = cp->pending_messages_head;
while ( (pos != NULL) &&
(pm->priority < pos->priority) )
}
-/**
- * Mingle hash with the mingle_number to produce different bits.
- */
-static void
-mingle_hash (const GNUNET_HashCode * in,
- int32_t mingle_number,
- GNUNET_HashCode * hc)
-{
- GNUNET_HashCode m;
-
- GNUNET_CRYPTO_hash (&mingle_number,
- sizeof (int32_t),
- &m);
- GNUNET_CRYPTO_hash_xor (&m, in, hc);
-}
-
-
/**
* Test if the load on this peer is too high
* to even consider processing the query at
* all.
*
- * @return GNUNET_YES if the load is too high, GNUNET_NO otherwise
+ * @return GNUNET_YES if the load is too high to do anything, GNUNET_NO to forward (load high, but not too high), GNUNET_SYSERR to indirect (load low)
*/
static int
test_load_too_high ()
{
- return GNUNET_NO; // FIXME
+ return GNUNET_SYSERR; // FIXME
}
BLOOMFILTER_K);
for (i=0;i<pr->replies_seen_off;i++)
{
- mingle_hash (&pr->replies_seen[i], pr->mingle, &mhash);
+ GNUNET_BLOCK_mingle_hash (&pr->replies_seen[i],
+ pr->mingle,
+ &mhash);
GNUNET_CONTAINER_bloomfilter_add (pr->bf, &mhash);
}
}
/* 3e) include peer proximity */
score -= (2.0 * (GNUNET_CRYPTO_hash_distance_u32 (key,
&pr->query)) / (double) UINT32_MAX);
+ /* 4) super-bonus for being the known target */
+ if (pr->target_pid == cp->pid)
+ score += 100.0;
/* store best-fit in closure */
#if DEBUG_FS
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
}
if (GNUNET_YES == pr->local_only)
return; /* configured to not do P2P search */
+ /* (0) try DHT */
+ if (0 == pr->anonymity_level)
+ {
+#if 0
+ /* DHT API needs fixing... */
+ pr->dht_get = GNUNET_DHT_get_start (dht_handle,
+ GNUNET_TIME_UNIT_FOREVER_REL,
+ pr->type,
+ &pr->query,
+ &process_dht_reply,
+ pr,
+ FIXME,
+ FIXME);
+#endif
+ }
/* (1) select target */
psc.pr = pr;
psc.target_score = -DBL_MAX;
if (0 == GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
4))
pr->priority++;
- /* FIXME: bound priority by "customary" priority used by other peers
- at this time! */
+ /* bound priority we use by priorities we see from other peers
+ rounded up (must round up so that we can see non-zero
+ priorities, but round up as little as possible to make it
+ plausible that we forwarded another peers request) */
+ if (pr->priority > current_priorities + 1.0)
+ pr->priority = (uint32_t) current_priorities + 1.0;
pr->ttl = bound_ttl (pr->ttl + TTL_DECREMENT * 2,
pr->priority);
#if DEBUG_FS
switch (pr->type)
{
- case GNUNET_BLOCK_TYPE_DBLOCK:
- case GNUNET_BLOCK_TYPE_IBLOCK:
+ case GNUNET_BLOCK_TYPE_FS_DBLOCK:
+ case GNUNET_BLOCK_TYPE_FS_IBLOCK:
/* only one reply expected, done with the request! */
destroy_pending_request (pr);
break;
case GNUNET_BLOCK_TYPE_ANY:
- case GNUNET_BLOCK_TYPE_KBLOCK:
- case GNUNET_BLOCK_TYPE_SBLOCK:
+ case GNUNET_BLOCK_TYPE_FS_KBLOCK:
+ case GNUNET_BLOCK_TYPE_FS_SBLOCK:
break;
default:
GNUNET_break (0);
*/
size_t size;
- /**
- * Namespace that this reply belongs to
- * (if it is of type SBLOCK).
- */
- GNUNET_HashCode namespace;
-
/**
* Type of the block.
*/
*/
uint32_t priority;
+ /**
+ * Evaluation result (returned).
+ */
+ enum GNUNET_BLOCK_EvaluationResult eval;
+
/**
* Did we finish processing the associated request?
*/
int finished;
+
+ /**
+ * Did we find a matching request?
+ */
+ int request_found;
};
struct PutMessage *pm;
struct ConnectedPeer *cp;
struct GNUNET_TIME_Relative cur_delay;
- GNUNET_HashCode chash;
- GNUNET_HashCode mhash;
size_t msize;
#if DEBUG_FS
GNUNET_SERVER_client_keep (pr->client_request_list->client_list->client);
}
}
- GNUNET_CRYPTO_hash (prq->data,
- prq->size,
- &chash);
- switch (prq->type)
- {
- case GNUNET_BLOCK_TYPE_DBLOCK:
- case GNUNET_BLOCK_TYPE_IBLOCK:
- /* only possible reply, stop requesting! */
+ prq->eval = GNUNET_BLOCK_evaluate (block_ctx,
+ prq->type,
+ key,
+ &pr->bf,
+ pr->mingle,
+ pr->namespace, (pr->namespace != NULL) ? sizeof (GNUNET_HashCode) : 0,
+ prq->data,
+ prq->size);
+ switch (prq->eval)
+ {
+ case GNUNET_BLOCK_EVALUATION_OK_MORE:
+ break;
+ case GNUNET_BLOCK_EVALUATION_OK_LAST:
while (NULL != pr->pending_head)
destroy_pending_message_list_entry (pr->pending_head);
if (pr->qe != NULL)
if (pr->client_request_list != NULL)
GNUNET_SERVER_receive_done (pr->client_request_list->client_list->client,
GNUNET_YES);
- GNUNET_DATASTORE_cancel (pr->qe);
+ GNUNET_DATASTORE_cancel (pr->qe);
pr->qe = NULL;
}
pr->do_remove = GNUNET_YES;
key,
pr));
break;
- case GNUNET_BLOCK_TYPE_SBLOCK:
- if (pr->namespace == NULL)
- {
- GNUNET_break (0);
- return GNUNET_YES;
- }
- if (0 != memcmp (pr->namespace,
- &prq->namespace,
- sizeof (GNUNET_HashCode)))
- {
- GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
- _("Reply mismatched in terms of namespace. Discarded.\n"));
- return GNUNET_YES; /* wrong namespace */
- }
- /* then: fall-through! */
- case GNUNET_BLOCK_TYPE_KBLOCK:
- case GNUNET_BLOCK_TYPE_NBLOCK:
- if (pr->bf != NULL)
- {
- mingle_hash (&chash, pr->mingle, &mhash);
- if (GNUNET_YES == GNUNET_CONTAINER_bloomfilter_test (pr->bf,
- &mhash))
- {
- GNUNET_STATISTICS_update (stats,
- gettext_noop ("# duplicate replies discarded (bloomfilter)"),
- 1,
- GNUNET_NO);
-#if DEBUG_FS
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Duplicate response `%s', discarding.\n",
- GNUNET_h2s (&mhash));
-#endif
- return GNUNET_YES; /* duplicate */
- }
+ case GNUNET_BLOCK_EVALUATION_OK_DUPLICATE:
+ GNUNET_STATISTICS_update (stats,
+ gettext_noop ("# duplicate replies discarded (bloomfilter)"),
+ 1,
+ GNUNET_NO);
#if DEBUG_FS
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "New response `%s', adding to filter.\n",
- GNUNET_h2s (&mhash));
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Duplicate response `%s', discarding.\n",
+ GNUNET_h2s (&mhash));
#endif
- }
- if (pr->client_request_list != NULL)
- {
- if (pr->replies_seen_size == pr->replies_seen_off)
- GNUNET_array_grow (pr->replies_seen,
- pr->replies_seen_size,
- pr->replies_seen_size * 2 + 4);
- pr->replies_seen[pr->replies_seen_off++] = chash;
- }
- if ( (pr->bf == NULL) ||
- (pr->client_request_list != NULL) )
- refresh_bloomfilter (pr);
- GNUNET_CONTAINER_bloomfilter_add (pr->bf,
- &mhash);
- break;
- default:
+ return GNUNET_YES; /* duplicate */
+ case GNUNET_BLOCK_EVALUATION_RESULT_INVALID:
+ return GNUNET_YES; /* wrong namespace */
+ case GNUNET_BLOCK_EVALUATION_REQUEST_VALID:
+ GNUNET_break (0);
+ return GNUNET_YES;
+ case GNUNET_BLOCK_EVALUATION_REQUEST_INVALID:
GNUNET_break (0);
return GNUNET_YES;
+ case GNUNET_BLOCK_EVALUATION_TYPE_NOT_SUPPORTED:
+ GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
+ _("Unsupported block type %u\n"),
+ prq->type);
+ return GNUNET_NO;
+ }
+ if (pr->client_request_list != NULL)
+ {
+ if (pr->replies_seen_size == pr->replies_seen_off)
+ GNUNET_array_grow (pr->replies_seen,
+ pr->replies_seen_size,
+ pr->replies_seen_size * 2 + 4);
+ GNUNET_CRYPTO_hash (prq->data,
+ prq->size,
+ &pr->replies_seen[pr->replies_seen_off++]);
+ refresh_bloomfilter (pr);
+ }
+ if (NULL == prq->sender)
+ {
+#if DEBUG_FS
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Found result for query `%s' in local datastore\n",
+ GNUNET_h2s (key));
+#endif
+ GNUNET_STATISTICS_update (stats,
+ gettext_noop ("# results found locally"),
+ 1,
+ GNUNET_NO);
}
prq->priority += pr->remaining_priority;
pr->remaining_priority = 0;
+ pr->results_found++;
+ prq->request_found = GNUNET_YES;
if (NULL != pr->client_request_list)
{
GNUNET_STATISTICS_update (stats,
int success,
const char *msg)
{
- /* FIXME */
+ struct GNUNET_TIME_Absolute *start = cls;
+ struct GNUNET_TIME_Relative delay;
+
+ delay = GNUNET_TIME_absolute_get_duration (*start);
+ GNUNET_free (start);
+ GNUNET_LOAD_update (datastore_put_load,
+ delay.value);
+ if (GNUNET_OK == success)
+ return;
+ GNUNET_STATISTICS_update (stats,
+ gettext_noop ("# datastore 'put' failures"),
+ 1,
+ GNUNET_NO);
}
struct GNUNET_TIME_Absolute expiration;
GNUNET_HashCode query;
struct ProcessReplyClosure prq;
- const struct SBlock *sb;
- struct ConnectedPeer *cps;
+ struct GNUNET_TIME_Absolute *start;
+ struct GNUNET_TIME_Relative block_time;
+ double putl;
+ struct ConnectedPeer *cp;
+ struct PendingMessage *pm;
+ struct MigrationStopMessage *msm;
msize = ntohs (message->size);
if (msize < sizeof (struct PutMessage))
type = ntohl (put->type);
expiration = GNUNET_TIME_absolute_ntoh (put->expiration);
+ if (type == GNUNET_BLOCK_TYPE_FS_ONDEMAND)
+ return GNUNET_SYSERR;
if (GNUNET_OK !=
- GNUNET_BLOCK_check_block (type,
- &put[1],
- dsize,
- &query))
+ GNUNET_BLOCK_get_key (block_ctx,
+ type,
+ &put[1],
+ dsize,
+ &query))
{
GNUNET_break_op (0);
return GNUNET_SYSERR;
}
- if (type == GNUNET_BLOCK_TYPE_ONDEMAND)
- return GNUNET_SYSERR;
- if (GNUNET_BLOCK_TYPE_SBLOCK == type)
- {
- sb = (const struct SBlock*) &put[1];
- GNUNET_CRYPTO_hash (&sb->subspace,
- sizeof (struct GNUNET_CRYPTO_RsaPublicKeyBinaryEncoded),
- &prq.namespace);
- }
-
#if DEBUG_FS
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
"Received result for query `%s' from peer `%4s'\n",
if (other != NULL)
prq.sender = GNUNET_CONTAINER_multihashmap_get (connected_peers,
&other->hashPubKey);
+ else
+ prq.sender = NULL;
prq.size = dsize;
prq.type = type;
prq.expiration = expiration;
prq.priority = 0;
prq.finished = GNUNET_NO;
+ prq.request_found = GNUNET_NO;
GNUNET_CONTAINER_multihashmap_get_multiple (query_request_map,
&query,
&process_reply,
&prq);
- cps = GNUNET_CONTAINER_multihashmap_get (connected_peers,
- &other->hashPubKey);
- if (cps != NULL)
+ if (prq.sender != NULL)
{
- cps->inc_preference += CONTENT_BANDWIDTH_VALUE + 1000 * prq.priority;
- cps->trust_delta += prq.priority;
- }
- else
- {
- GNUNET_break (0);
+ prq.sender->inc_preference += CONTENT_BANDWIDTH_VALUE + 1000 * prq.priority;
+ prq.sender->trust += prq.priority;
}
if (GNUNET_YES == active_migration)
{
GNUNET_h2s (&query),
prq.priority);
#endif
+ start = GNUNET_malloc (sizeof (struct GNUNET_TIME_Absolute));
+ *start = GNUNET_TIME_absolute_get ();
GNUNET_DATASTORE_put (dsh,
0, &query, dsize, &put[1],
type, prq.priority, 1 /* anonymity */,
1 + prq.priority, MAX_DATASTORE_QUEUE,
GNUNET_CONSTANTS_SERVICE_TIMEOUT,
&put_migration_continuation,
- NULL);
+ start);
+ }
+ putl = GNUNET_LOAD_get_load (datastore_put_load);
+ if ( (GNUNET_NO == prq.request_found) &&
+ ( (GNUNET_YES != active_migration) ||
+ (putl > 2.0) ) )
+ {
+ cp = GNUNET_CONTAINER_multihashmap_get (connected_peers,
+ &other->hashPubKey);
+ if (GNUNET_TIME_absolute_get_duration (cp->last_migration_block).value < 5000)
+ return GNUNET_OK; /* already blocked */
+ /* We're too busy; send MigrationStop message! */
+ if (GNUNET_YES != active_migration)
+ putl = 1.0 + GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK, 5);
+ block_time = GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MILLISECONDS,
+ 5000 + GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
+ (unsigned int) (60000 * putl * putl)));
+
+ cp->last_migration_block = GNUNET_TIME_relative_to_absolute (block_time);
+ pm = GNUNET_malloc (sizeof (struct PendingMessage) +
+ sizeof (struct MigrationStopMessage));
+ pm->msize = sizeof (struct MigrationStopMessage);
+ pm->priority = UINT32_MAX;
+ msm = (struct MigrationStopMessage*) &pm[1];
+ msm->header.size = htons (sizeof (struct MigrationStopMessage));
+ msm->header.type = htons (GNUNET_MESSAGE_TYPE_FS_MIGRATION_STOP);
+ msm->duration = GNUNET_TIME_relative_hton (block_time);
+ add_to_pending_messages_for_peer (cp,
+ pm,
+ NULL);
+ }
+ return GNUNET_OK;
+}
+
+
+/**
+ * Handle P2P "MIGRATION_STOP" message.
+ *
+ * @param cls closure, always NULL
+ * @param other the other peer involved (sender or receiver, NULL
+ * for loopback messages where we are both sender and receiver)
+ * @param message the actual message
+ * @param latency reported latency of the connection with 'other'
+ * @param distance reported distance (DV) to 'other'
+ * @return GNUNET_OK to keep the connection open,
+ * GNUNET_SYSERR to close it (signal serious error)
+ */
+static int
+handle_p2p_migration_stop (void *cls,
+ const struct GNUNET_PeerIdentity *other,
+ const struct GNUNET_MessageHeader *message,
+ struct GNUNET_TIME_Relative latency,
+ uint32_t distance)
+{
+ struct ConnectedPeer *cp;
+ const struct MigrationStopMessage *msm;
+
+ msm = (const struct MigrationStopMessage*) message;
+ cp = GNUNET_CONTAINER_multihashmap_get (connected_peers,
+ &other->hashPubKey);
+ if (cp == NULL)
+ {
+ GNUNET_break (0);
+ return GNUNET_OK;
}
+ cp->migration_blocked = GNUNET_TIME_relative_to_absolute (GNUNET_TIME_relative_ntoh (msm->duration));
return GNUNET_OK;
}
+
/* **************************** P2P GET Handling ************************ */
struct PendingRequest *pr = cls;
struct ProcessReplyClosure prq;
struct CheckDuplicateRequestClosure cdrc;
- const struct SBlock *sb;
- GNUNET_HashCode dhash;
- GNUNET_HashCode mhash;
GNUNET_HashCode query;
+ unsigned int old_rf;
if (NULL == key)
{
GNUNET_h2s (key),
type);
#endif
- if (type == GNUNET_BLOCK_TYPE_ONDEMAND)
+ if (type == GNUNET_BLOCK_TYPE_FS_ONDEMAND)
{
#if DEBUG_FS
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
&process_local_reply,
pr))
if (pr->qe != NULL)
- GNUNET_DATASTORE_get_next (dsh, GNUNET_YES);
- return;
- }
- /* check for duplicates */
- GNUNET_CRYPTO_hash (data, size, &dhash);
- mingle_hash (&dhash,
- pr->mingle,
- &mhash);
- if ( (pr->bf != NULL) &&
- (GNUNET_YES ==
- GNUNET_CONTAINER_bloomfilter_test (pr->bf,
- &mhash)) )
- {
-#if DEBUG_FS
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Result from datastore filtered by bloomfilter (duplicate).\n");
-#endif
- GNUNET_STATISTICS_update (stats,
- gettext_noop ("# results filtered by query bloomfilter"),
- 1,
- GNUNET_NO);
- if (pr->qe != NULL)
- GNUNET_DATASTORE_get_next (dsh, GNUNET_YES);
+ {
+ GNUNET_DATASTORE_get_next (dsh, GNUNET_YES);
+ }
return;
}
-#if DEBUG_FS
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Found result for query `%s' in local datastore\n",
- GNUNET_h2s (key));
-#endif
- GNUNET_STATISTICS_update (stats,
- gettext_noop ("# results found locally"),
- 1,
- GNUNET_NO);
- pr->results_found++;
+ old_rf = pr->results_found;
memset (&prq, 0, sizeof (prq));
prq.data = data;
prq.expiration = expiration;
prq.size = size;
- if (GNUNET_BLOCK_TYPE_SBLOCK == type)
- {
- sb = (const struct SBlock*) data;
- GNUNET_CRYPTO_hash (&sb->subspace,
- sizeof (struct GNUNET_CRYPTO_RsaPublicKeyBinaryEncoded),
- &prq.namespace);
- }
- if (GNUNET_OK != GNUNET_BLOCK_check_block (type,
- data,
- size,
- &query))
+ if (GNUNET_OK !=
+ GNUNET_BLOCK_get_key (block_ctx,
+ type,
+ data,
+ size,
+ &query))
{
GNUNET_break (0);
GNUNET_DATASTORE_remove (dsh,
prq.type = type;
prq.priority = priority;
prq.finished = GNUNET_NO;
+ prq.request_found = GNUNET_NO;
process_reply (&prq, key, pr);
+ if ( (old_rf == 0) &&
+ (pr->results_found == 1) )
+ update_datastore_delays (pr->start_time);
if (prq.finished == GNUNET_YES)
return;
if (pr->qe == NULL)
return; /* done here */
- if ( (type == GNUNET_BLOCK_TYPE_DBLOCK) ||
- (type == GNUNET_BLOCK_TYPE_IBLOCK) )
+ if (prq.eval == GNUNET_BLOCK_EVALUATION_OK_LAST)
{
GNUNET_DATASTORE_get_next (dsh, GNUNET_NO);
return;
bound_priority (uint32_t prio_in,
struct ConnectedPeer *cp)
{
- if (cp->trust_delta > prio_in)
- {
- cp->trust_delta -= prio_in;
- return prio_in;
+#define N ((double)128.0)
+ uint32_t ret;
+ double rret;
+ int ld;
+
+ ld = test_load_too_high ();
+ if (ld == GNUNET_SYSERR)
+ return 0; /* excess resources */
+ ret = change_host_trust (cp, prio_in);
+ if (ret > 0)
+ {
+ if (ret > current_priorities + N)
+ rret = current_priorities + N;
+ else
+ rret = ret;
+ current_priorities
+ = (current_priorities * (N-1) + rret)/N;
}
- // FIXME: get out trust in the target peer from peerinfo!
- return 0;
+#undef N
+ return ret;
}
uint32_t ttl_decrement;
enum GNUNET_BLOCK_Type type;
int have_ns;
+ int ld;
msize = ntohs(message->size);
if (msize < sizeof (struct GetMessage))
}
gm = (const struct GetMessage*) message;
type = ntohl (gm->type);
- switch (type)
- {
- case GNUNET_BLOCK_TYPE_ANY:
- case GNUNET_BLOCK_TYPE_DBLOCK:
- case GNUNET_BLOCK_TYPE_IBLOCK:
- case GNUNET_BLOCK_TYPE_KBLOCK:
- case GNUNET_BLOCK_TYPE_SBLOCK:
- break;
- default:
- GNUNET_break_op (0);
- return GNUNET_SYSERR;
- }
bm = ntohl (gm->hash_bitmap);
bits = 0;
while (bm > 0)
opt = (const GNUNET_HashCode*) &gm[1];
bfsize = msize - sizeof (struct GetMessage) + bits * sizeof (GNUNET_HashCode);
bm = ntohl (gm->hash_bitmap);
- if ( (0 != (bm & GET_MESSAGE_BIT_SKS_NAMESPACE)) &&
- (type != GNUNET_BLOCK_TYPE_SBLOCK) )
- {
- GNUNET_break_op (0);
- return GNUNET_SYSERR;
- }
bits = 0;
cps = GNUNET_CONTAINER_multihashmap_get (connected_peers,
&other->hashPubKey);
/* 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... */
- if (GNUNET_YES == test_load_too_high ())
+
+ /* FIXME: query priority should play
+ a major role here! */
+ ld = test_load_too_high ();
+ if (GNUNET_YES == ld)
{
#if DEBUG_FS
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
GNUNET_NO);
return GNUNET_OK;
}
+ /* FIXME: if ld == GNUNET_NO, forward
+ instead of indirecting! */
#if DEBUG_FS
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
pr->mingle = ntohl (gm->filter_mutator);
if (0 != (bm & GET_MESSAGE_BIT_TRANSMIT_TO))
pr->target_pid = GNUNET_PEER_intern ((const struct GNUNET_PeerIdentity*) &opt[bits++]);
-
pr->anonymity_level = 1;
pr->priority = bound_priority (ntohl (gm->priority), cps);
pr->ttl = bound_ttl (ntohl (gm->ttl), pr->priority);
/* calculate change in traffic preference */
cps->inc_preference += pr->priority * 1000 + QUERY_BANDWIDTH_VALUE;
/* process locally */
- if (type == GNUNET_BLOCK_TYPE_DBLOCK)
+ if (type == GNUNET_BLOCK_TYPE_FS_DBLOCK)
type = GNUNET_BLOCK_TYPE_ANY; /* to get on-demand as well */
timeout = GNUNET_TIME_relative_multiply (BASIC_DATASTORE_REQUEST_DELAY,
(pr->priority + 1));
/* Are multiple results possible? If so, start processing remotely now! */
switch (pr->type)
{
- case GNUNET_BLOCK_TYPE_DBLOCK:
- case GNUNET_BLOCK_TYPE_IBLOCK:
+ case GNUNET_BLOCK_TYPE_FS_DBLOCK:
+ case GNUNET_BLOCK_TYPE_FS_IBLOCK:
/* only one result, wait for datastore */
break;
default:
if (GNUNET_CONTAINER_heap_get_size (requests_by_expiration_heap) > max_pending_requests)
{
pr = GNUNET_CONTAINER_heap_peek (requests_by_expiration_heap);
+ GNUNET_assert (pr != NULL);
destroy_pending_request (pr);
}
return GNUNET_OK;
GNUNET_h2s (&sm->query),
(unsigned int) type);
#endif
- switch (type)
- {
- case GNUNET_BLOCK_TYPE_ANY:
- case GNUNET_BLOCK_TYPE_DBLOCK:
- case GNUNET_BLOCK_TYPE_IBLOCK:
- case GNUNET_BLOCK_TYPE_KBLOCK:
- case GNUNET_BLOCK_TYPE_SBLOCK:
- case GNUNET_BLOCK_TYPE_NBLOCK:
- break;
- default:
- GNUNET_break (0);
- GNUNET_SERVER_receive_done (client,
- GNUNET_SYSERR);
- return;
- }
-
cl = client_list;
while ( (cl != NULL) &&
(cl->client != client) )
client_list = cl;
}
/* detect duplicate KBLOCK requests */
- if ( (type == GNUNET_BLOCK_TYPE_KBLOCK) ||
- (type == GNUNET_BLOCK_TYPE_NBLOCK) ||
+ if ( (type == GNUNET_BLOCK_TYPE_FS_KBLOCK) ||
+ (type == GNUNET_BLOCK_TYPE_FS_NBLOCK) ||
(type == GNUNET_BLOCK_TYPE_ANY) )
{
crl = cl->rl_head;
1,
GNUNET_NO);
pr = GNUNET_malloc (sizeof (struct PendingRequest) +
- ((type == GNUNET_BLOCK_TYPE_SBLOCK) ? sizeof(GNUNET_HashCode) : 0));
+ ((type == GNUNET_BLOCK_TYPE_FS_SBLOCK) ? sizeof(GNUNET_HashCode) : 0));
crl = GNUNET_malloc (sizeof (struct ClientRequestList));
memset (crl, 0, sizeof (struct ClientRequestList));
crl->client_list = cl;
sc * sizeof (GNUNET_HashCode));
pr->replies_seen_off = sc;
pr->anonymity_level = ntohl (sm->anonymity_level);
+ pr->start_time = GNUNET_TIME_absolute_get ();
refresh_bloomfilter (pr);
pr->query = sm->query;
if (0 == (1 & ntohl (sm->options)))
pr->local_only = GNUNET_YES;
switch (type)
{
- case GNUNET_BLOCK_TYPE_DBLOCK:
- case GNUNET_BLOCK_TYPE_IBLOCK:
+ case GNUNET_BLOCK_TYPE_FS_DBLOCK:
+ case GNUNET_BLOCK_TYPE_FS_IBLOCK:
if (0 != memcmp (&sm->target,
&all_zeros,
sizeof (GNUNET_HashCode)))
pr->target_pid = GNUNET_PEER_intern ((const struct GNUNET_PeerIdentity*) &sm->target);
break;
- case GNUNET_BLOCK_TYPE_SBLOCK:
+ case GNUNET_BLOCK_TYPE_FS_SBLOCK:
pr->namespace = (GNUNET_HashCode*) &pr[1];
memcpy (&pr[1], &sm->target, sizeof (GNUNET_HashCode));
break;
&sm->query,
pr,
GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE));
- if (type == GNUNET_BLOCK_TYPE_DBLOCK)
+ if (type == GNUNET_BLOCK_TYPE_FS_DBLOCK)
type = GNUNET_BLOCK_TYPE_ANY; /* get on-demand blocks too! */
pr->qe = GNUNET_DATASTORE_get (dsh,
&sm->query,
GNUNET_MESSAGE_TYPE_FS_GET, 0 },
{ &handle_p2p_put,
GNUNET_MESSAGE_TYPE_FS_PUT, 0 },
+ { &handle_p2p_migration_stop,
+ GNUNET_MESSAGE_TYPE_FS_MIGRATION_STOP,
+ sizeof (struct MigrationStopMessage) },
{ NULL, 0, 0 }
};
static const struct GNUNET_SERVER_MessageHandler handlers[] = {
0 },
{NULL, NULL, 0, 0}
};
+ unsigned long long enc = 128;
sched = s;
cfg = c;
stats = GNUNET_STATISTICS_create (sched, "fs", cfg);
- min_migration_delay = GNUNET_TIME_UNIT_SECONDS; // FIXME: get from config
- connected_peers = GNUNET_CONTAINER_multihashmap_create (128); // FIXME: get size from config
- query_request_map = GNUNET_CONTAINER_multihashmap_create (128); // FIXME: get size from config
- peer_request_map = GNUNET_CONTAINER_multihashmap_create (128); // FIXME: get size from config
+ min_migration_delay = GNUNET_TIME_UNIT_SECONDS;
+ if ( (GNUNET_OK !=
+ GNUNET_CONFIGURATION_get_value_number (cfg,
+ "fs",
+ "MAX_PENDING_REQUESTS",
+ &max_pending_requests)) ||
+ (GNUNET_OK !=
+ GNUNET_CONFIGURATION_get_value_number (cfg,
+ "fs",
+ "EXPECTED_NEIGHBOUR_COUNT",
+ &enc)) ||
+ (GNUNET_OK !=
+ GNUNET_CONFIGURATION_get_value_time (cfg,
+ "fs",
+ "MIN_MIGRATION_DELAY",
+ &min_migration_delay)) )
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+ _("Configuration fails to specify certain parameters, assuming default values."));
+ }
+ connected_peers = GNUNET_CONTAINER_multihashmap_create (enc);
+ query_request_map = GNUNET_CONTAINER_multihashmap_create (max_pending_requests);
+ peer_request_map = GNUNET_CONTAINER_multihashmap_create (enc);
requests_by_expiration_heap = GNUNET_CONTAINER_heap_create (GNUNET_CONTAINER_HEAP_ORDER_MIN);
core = GNUNET_CORE_connect (sched,
cfg,
NULL,
&peer_connect_handler,
&peer_disconnect_handler,
+ NULL,
NULL, GNUNET_NO,
NULL, GNUNET_NO,
p2p_handlers);
GNUNET_SERVER_disconnect_notify (server,
&handle_client_disconnect,
NULL);
+ GNUNET_assert (GNUNET_OK ==
+ GNUNET_CONFIGURATION_get_value_filename (cfg,
+ "fs",
+ "TRUST",
+ &trustDirectory));
+ GNUNET_DISK_directory_create (trustDirectory);
+ GNUNET_SCHEDULER_add_with_priority (sched,
+ GNUNET_SCHEDULER_PRIORITY_HIGH,
+ &cron_flush_trust, NULL);
+
+
GNUNET_SERVER_add_handlers (server, handlers);
GNUNET_SCHEDULER_add_delayed (sched,
GNUNET_TIME_UNIT_FOREVER_REL,
GNUNET_SCHEDULER_shutdown (sched);
return;
}
+ datastore_get_load = GNUNET_LOAD_value_init ();
+ datastore_put_load = GNUNET_LOAD_value_init ();
+ block_cfg = GNUNET_CONFIGURATION_create ();
+ GNUNET_CONFIGURATION_set_value_string (block_cfg,
+ "block",
+ "PLUGINS",
+ "fs");
+ block_ctx = GNUNET_BLOCK_context_create (block_cfg);
+ GNUNET_assert (NULL != block_ctx);
+ dht_handle = GNUNET_DHT_connect (sched,
+ cfg,
+ FS_DHT_HT_SIZE);
if ( (GNUNET_OK != GNUNET_FS_indexing_init (sched, cfg, dsh)) ||
(GNUNET_OK != main_init (sched, server, cfg)) )
{
GNUNET_SCHEDULER_shutdown (sched);
GNUNET_DATASTORE_disconnect (dsh, GNUNET_NO);
dsh = NULL;
+ GNUNET_DHT_disconnect (dht_handle);
+ dht_handle = NULL;
+ GNUNET_BLOCK_context_destroy (block_ctx);
+ block_ctx = NULL;
+ GNUNET_CONFIGURATION_destroy (block_cfg);
+ block_cfg = NULL;
+ GNUNET_LOAD_value_free (datastore_get_load);
+ datastore_get_load = NULL;
+ GNUNET_LOAD_value_free (datastore_put_load);
+ datastore_put_load = NULL;
return;
}
}