* @brief gnunet anonymity protocol implementation
* @author Christian Grothoff
*
- * FIXME:
- * - TTL/priority calculations are absent!
* TODO:
- * - have non-zero preference / priority for requests we initiate!
- * - implement hot-path routing decision procedure
- * - implement: bound_priority, test_load_too_high
- * - statistics
+ * - 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
+ * - 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>
/**
* Maximum number of outgoing messages we queue per peer.
- * FIXME: make configurable?
*/
#define MAX_QUEUE_PER_PEER 16
/**
* Increase in traffic preference still to be submitted
- * to the core service for this peer. FIXME: double or 'uint64_t'?
+ * to the core service for this peer.
*/
- double inc_preference;
+ uint64_t inc_preference;
+
+ /**
+ * Trust delta to still commit to the system.
+ */
+ uint32_t trust_delta;
/**
* The peer's identity.
struct GNUNET_DATASTORE_QueueEntry *qe;
/**
-
* Size of the 'bf' (in bytes).
*/
size_t bf_size;
* targets for (or NULL for none)
* @param key ID of the peer
* @param value 'struct ConnectedPeer' of the peer
- * @return GNUNET_YES (always continue iteration)2
+ * @return GNUNET_YES (always continue iteration)
*/
static int
consider_migration (void *cls,
}
if (msize == 0)
return GNUNET_YES; /* no content available */
+#if DEBUG_FS
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Trying to migrate at least %u bytes to peer `%s'\n",
+ msize,
+ GNUNET_h2s (key));
+#endif
cp->cth
= GNUNET_CORE_notify_transmit_ready (core,
0, GNUNET_TIME_UNIT_FOREVER_REL,
return;
delay = GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS,
mig_size);
- delay = GNUNET_TIME_relative_divide (GNUNET_TIME_UNIT_SECONDS,
+ delay = GNUNET_TIME_relative_divide (delay,
MAX_MIGRATION_QUEUE);
delay = GNUNET_TIME_relative_max (delay,
min_migration_delay);
GNUNET_DATASTORE_get_next (dsh, GNUNET_YES);
return;
}
+#if DEBUG_FS
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Retrieved block `%s' of type %u for migration\n",
+ GNUNET_h2s (key),
+ type);
+#endif
mb = GNUNET_malloc (sizeof (struct MigrationReadyBlock) + size);
mb->query = *key;
mb->expiration = expiration;
{
GNUNET_PEER_change_rc (pos->target_list[i], -1);
pos->target_list[i] = 0;
- if (pos->used_targets >= GNUNET_CONTAINER_multihashmap_size (connected_peers))
- {
- delete_migration_block (pos);
- consider_migration_gathering ();
- }
- GNUNET_CONTAINER_multihashmap_iterate (connected_peers,
- &consider_migration,
- pos);
- break;
- }
+ }
+ }
+ if (pos->used_targets >= GNUNET_CONTAINER_multihashmap_size (connected_peers))
+ {
+ delete_migration_block (pos);
+ consider_migration_gathering ();
+ continue;
}
+ GNUNET_CONTAINER_multihashmap_iterate (connected_peers,
+ &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)
size -= sizeof (migm);
memcpy (&cbuf[msize], &mb[1], mb->size);
msize += mb->size;
- size -= mb->size;
+ size -= mb->size;
+#if DEBUG_FS
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Pushing migration block `%s' (%u bytes) to `%s'\n",
+ GNUNET_h2s (&mb->query),
+ mb->size,
+ GNUNET_i2s (&pid));
+#endif
break;
}
+ else
+ {
+#if DEBUG_FS
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Migration block `%s' (%u bytes) is not on migration list for peer `%s'\n",
+ GNUNET_h2s (&mb->query),
+ mb->size,
+ GNUNET_i2s (&pid));
+#endif
+ }
}
if ( (mb->used_targets >= MIGRATION_TARGET_COUNT) ||
(mb->used_targets >= GNUNET_CONTAINER_multihashmap_size (connected_peers)) )
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
"Failed to schedule transmission with core!\n");
#endif
- /* FIXME: call stats (rare, bad case) */
+ GNUNET_STATISTICS_update (stats,
+ gettext_noop ("# CORE transmission failures"),
+ 1,
+ GNUNET_NO);
}
}
return;
}
no_route = GNUNET_NO;
- /* FIXME: check against DBLOCK_SIZE and possibly return
- amount to reserve; however, this also needs to work
- with testcases which currently start out with a far
- too low per-peer bw limit, so they would never send
- anything. Big issue. */
if (amount == 0)
{
if (pr->cp == NULL)
/* 1) check that this peer is not the initiator */
if (cp == pr->cp)
- return GNUNET_YES; /* skip */
+ {
+#if DEBUG_FS
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Skipping initiator in forwarding selection\n");
+#endif
+ return GNUNET_YES; /* skip */
+ }
/* 2) check if we have already (recently) forwarded to this peer */
pc = 0;
"Re-trying query that was previously transmitted %u times to this peer\n",
(unsigned int) pc);
#endif
- // 3) calculate how much we'd like to forward to this peer
- score = 42; // FIXME!
- // FIXME: also need API to gather data on responsiveness
- // of this peer (we have fields for that in 'cp', but
- // they are never set!)
-
+ /* 3) calculate how much we'd like to forward to this peer,
+ starting with a random value that is strong enough
+ to at least give any peer a chance sometimes
+ (compared to the other factors that come later) */
+ /* 3a) count successful (recent) routes from cp for same source */
+ if (pr->cp != NULL)
+ {
+ score = GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
+ P2P_SUCCESS_LIST_SIZE);
+ for (i=0;i<P2P_SUCCESS_LIST_SIZE;i++)
+ if (cp->last_p2p_replies[i] == pr->cp->pid)
+ score += 1; /* likely successful based on hot path */
+ }
+ else
+ {
+ score = GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
+ CS2P_SUCCESS_LIST_SIZE);
+ for (i=0;i<CS2P_SUCCESS_LIST_SIZE;i++)
+ if (cp->last_client_replies[i] == pr->client_request_list->client_list->client)
+ score += 1; /* likely successful based on hot path */
+ }
+ /* 3b) include latency */
+ if (cp->avg_delay.value < 4 * TTL_DECREMENT)
+ score += 1; /* likely fast based on latency */
+ /* 3c) include priorities */
+ if (cp->avg_priority <= pr->remaining_priority / 2.0)
+ score += 1; /* likely successful based on priorities */
+ /* 3d) penalize for queue size */
+ score -= (2.0 * cp->pending_requests / (double) MAX_QUEUE_PER_PEER);
+ /* 3e) include peer proximity */
+ score -= (2.0 * (GNUNET_CRYPTO_hash_distance_u32 (key,
+ &pr->query)) / (double) UINT32_MAX);
/* store best-fit in closure */
+#if DEBUG_FS
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Peer `%s' gets score %f for forwarding query, max is %f\n",
+ GNUNET_h2s (key),
+ score,
+ psc->target_score);
+#endif
+ score++; /* avoid zero */
if (score > psc->target_score)
{
psc->target_score = score;
/**
- * We're processing a GET request from another peer and have decided
+ * We're processing a GET request and have decided
* to forward it to other peers. This function is called periodically
* and should forward the request to other peers until we have all
* possible replies. If we have transmitted the *only* reply to
return; /* configured to not do P2P search */
/* (1) select target */
psc.pr = pr;
- psc.target_score = DBL_MIN;
+ psc.target_score = -DBL_MAX;
GNUNET_CONTAINER_multihashmap_iterate (connected_peers,
&target_peer_select_cb,
&psc);
- if (psc.target_score == DBL_MIN)
+ if (psc.target_score == -DBL_MAX)
{
delay = get_processing_delay ();
#if DEBUG_FS
return; /* nobody selected */
}
/* (3) update TTL/priority */
-
if (pr->client_request_list != NULL)
{
/* FIXME: use better algorithm!? */
pr->ttl);
#endif
}
- else
- {
- /* FIXME: should we do something here as well!? */
- }
/* (3) reserve reply bandwidth */
cp = GNUNET_CONTAINER_multihashmap_get (connected_peers,
pr->irc = GNUNET_CORE_peer_change_preference (sched, cfg,
&psc.target,
GNUNET_CONSTANTS_SERVICE_TIMEOUT,
- GNUNET_BANDWIDTH_value_init ((uint32_t) -1 /* no limit */),
+ GNUNET_BANDWIDTH_value_init (UINT32_MAX),
DBLOCK_SIZE * 2,
- (uint64_t) cp->inc_preference,
+ cp->inc_preference,
&target_reservation_cb,
pr);
- cp->inc_preference = 0.0;
+ cp->inc_preference = 0;
}
* How much was this reply worth to us?
*/
uint32_t priority;
+
+ /**
+ * Did we finish processing the associated request?
+ */
+ int finished;
};
if (pr->cp != NULL)
{
GNUNET_PEER_change_rc (prq->sender->last_p2p_replies
- [prq->sender->last_p2p_replies_woff % P2P_SUCCESS_LIST_SIZE], -1);
+ [prq->sender->last_p2p_replies_woff % P2P_SUCCESS_LIST_SIZE],
+ -1);
GNUNET_PEER_change_rc (pr->cp->pid, 1);
prq->sender->last_p2p_replies
[(prq->sender->last_p2p_replies_woff++) % P2P_SUCCESS_LIST_SIZE]
}
prq->priority += pr->remaining_priority;
pr->remaining_priority = 0;
- if (pr->client_request_list != NULL)
+ if (NULL != pr->client_request_list)
{
GNUNET_STATISTICS_update (stats,
gettext_noop ("# replies received for local clients"),
}
GNUNET_break (cl->th != NULL);
if (pr->do_remove)
- destroy_pending_request (pr);
+ {
+ prq->finished = GNUNET_YES;
+ destroy_pending_request (pr);
+ }
}
else
{
reply->cont = &transmit_reply_continuation;
reply->cont_cls = pr;
reply->msize = msize;
- reply->priority = (uint32_t) -1; /* send replies first! */
+ reply->priority = UINT32_MAX; /* send replies first! */
pm = (struct PutMessage*) &reply[1];
pm->header.type = htons (GNUNET_MESSAGE_TYPE_FS_PUT);
pm->header.size = htons (msize);
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;
GNUNET_CONTAINER_multihashmap_get_multiple (query_request_map,
&query,
&process_reply,
&prq);
+ if (prq.sender != NULL)
+ {
+ prq.sender->inc_preference += CONTENT_BANDWIDTH_VALUE + 1000 * prq.priority;
+ prq.sender->trust_delta += prq.priority;
+ }
if (GNUNET_YES == active_migration)
{
+#if DEBUG_FS
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Replicating result for query `%s' with priority %u\n",
+ GNUNET_h2s (&query),
+ prq.priority);
+#endif
GNUNET_DATASTORE_put (dsh,
0, &query, dsize, &put[1],
type, prq.priority, 1 /* anonymity */,
GNUNET_DATASTORE_get_next (dsh, GNUNET_YES);
return;
}
+ prq.type = type;
+ prq.priority = priority;
+ prq.finished = GNUNET_NO;
+ process_reply (&prq, key, pr);
+ 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 (pr->qe != NULL)
- GNUNET_DATASTORE_get_next (dsh, GNUNET_NO);
+ GNUNET_DATASTORE_get_next (dsh, GNUNET_NO);
+ return;
}
- else if ( (pr->client_request_list == NULL) &&
+ if ( (pr->client_request_list == NULL) &&
( (GNUNET_YES == test_load_too_high()) ||
(pr->results_found > 5 + 2 * pr->priority) ) )
{
gettext_noop ("# processing result set cut short due to load"),
1,
GNUNET_NO);
- if (pr->qe != NULL)
- GNUNET_DATASTORE_get_next (dsh, GNUNET_NO);
+ GNUNET_DATASTORE_get_next (dsh, GNUNET_NO);
+ return;
}
- else if (pr->qe != NULL)
- GNUNET_DATASTORE_get_next (dsh, GNUNET_YES);
- prq.type = type;
- prq.priority = priority;
- process_reply (&prq, key, pr);
+ GNUNET_DATASTORE_get_next (dsh, GNUNET_YES);
}
bound_priority (uint32_t prio_in,
struct ConnectedPeer *cp)
{
- return 0; // FIXME!
+ if (cp->trust_delta > prio_in)
+ {
+ cp->trust_delta -= prio_in;
+ return prio_in;
+ }
+ // FIXME: get out trust in the target peer from peerinfo!
+ return 0;
}
size_t bfsize;
uint32_t ttl_decrement;
enum GNUNET_BLOCK_Type type;
- double preference;
int have_ns;
msize = ntohs(message->size);
pr = GNUNET_malloc (sizeof (struct PendingRequest) +
(have_ns ? sizeof(GNUNET_HashCode) : 0));
if (have_ns)
- pr->namespace = (GNUNET_HashCode*) &pr[1];
+ {
+ pr->namespace = (GNUNET_HashCode*) &pr[1];
+ memcpy (&pr[1], &opt[bits++], sizeof (GNUNET_HashCode));
+ }
pr->type = type;
pr->mingle = ntohl (gm->filter_mutator);
- if (0 != (bm & GET_MESSAGE_BIT_SKS_NAMESPACE))
- memcpy (&pr[1], &opt[bits++], sizeof (GNUNET_HashCode));
if (0 != (bm & GET_MESSAGE_BIT_TRANSMIT_TO))
pr->target_pid = GNUNET_PEER_intern ((const struct GNUNET_PeerIdentity*) &opt[bits++]);
gettext_noop ("# requests dropped due TTL underflow"),
1,
GNUNET_NO);
- /* integer underflow => drop (should be very rare)! */
+ /* integer underflow => drop (should be very rare)! */
GNUNET_free (pr);
return GNUNET_OK;
}
GNUNET_NO);
/* calculate change in traffic preference */
- preference = (double) pr->priority;
- if (preference < QUERY_BANDWIDTH_VALUE)
- preference = QUERY_BANDWIDTH_VALUE;
- cps->inc_preference += preference;
-
+ cps->inc_preference += pr->priority * 1000 + QUERY_BANDWIDTH_VALUE;
/* process locally */
if (type == GNUNET_BLOCK_TYPE_DBLOCK)
type = GNUNET_BLOCK_TYPE_ANY; /* to get on-demand as well */
/* **************************** Startup ************************ */
-
-/**
- * List of handlers for P2P messages
- * that we care about.
- */
-static struct GNUNET_CORE_MessageHandler p2p_handlers[] =
- {
- { &handle_p2p_get,
- GNUNET_MESSAGE_TYPE_FS_GET, 0 },
- { &handle_p2p_put,
- GNUNET_MESSAGE_TYPE_FS_PUT, 0 },
- { NULL, 0, 0 }
- };
-
-
-/**
- * List of handlers for the messages understood by this
- * service.
- */
-static struct GNUNET_SERVER_MessageHandler handlers[] = {
- {&GNUNET_FS_handle_index_start, NULL,
- GNUNET_MESSAGE_TYPE_FS_INDEX_START, 0},
- {&GNUNET_FS_handle_index_list_get, NULL,
- GNUNET_MESSAGE_TYPE_FS_INDEX_LIST_GET, sizeof(struct GNUNET_MessageHeader) },
- {&GNUNET_FS_handle_unindex, NULL, GNUNET_MESSAGE_TYPE_FS_UNINDEX,
- sizeof (struct UnindexMessage) },
- {&handle_start_search, NULL, GNUNET_MESSAGE_TYPE_FS_START_SEARCH,
- 0 },
- {NULL, NULL, 0, 0}
-};
-
-
/**
* Process fs requests.
*
struct GNUNET_SERVER_Handle *server,
const struct GNUNET_CONFIGURATION_Handle *c)
{
+ static const struct GNUNET_CORE_MessageHandler p2p_handlers[] =
+ {
+ { &handle_p2p_get,
+ GNUNET_MESSAGE_TYPE_FS_GET, 0 },
+ { &handle_p2p_put,
+ GNUNET_MESSAGE_TYPE_FS_PUT, 0 },
+ { NULL, 0, 0 }
+ };
+ static const struct GNUNET_SERVER_MessageHandler handlers[] = {
+ {&GNUNET_FS_handle_index_start, NULL,
+ GNUNET_MESSAGE_TYPE_FS_INDEX_START, 0},
+ {&GNUNET_FS_handle_index_list_get, NULL,
+ GNUNET_MESSAGE_TYPE_FS_INDEX_LIST_GET, sizeof(struct GNUNET_MessageHeader) },
+ {&GNUNET_FS_handle_unindex, NULL, GNUNET_MESSAGE_TYPE_FS_UNINDEX,
+ sizeof (struct UnindexMessage) },
+ {&handle_start_search, NULL, GNUNET_MESSAGE_TYPE_FS_START_SEARCH,
+ 0 },
+ {NULL, NULL, 0, 0}
+ };
+
sched = s;
cfg = c;
stats = GNUNET_STATISTICS_create (sched, "fs", cfg);
}
/* FIXME: distinguish between sending and storing in options? */
if (active_migration)
- consider_migration_gathering ();
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+ _("Content migration is enabled, will start to gather data\n"));
+ consider_migration_gathering ();
+ }
GNUNET_SERVER_disconnect_notify (server,
&handle_client_disconnect,
NULL);
projects / oweals / gnunet.git / blobdiff