*/
struct GNUNET_PeerIdentity id;
+ /**
+ *
+ */
struct Trail *pred_head;
+ /**
+ *
+ */
struct Trail *pred_tail;
+ /**
+ *
+ */
struct Trail *succ_head;
+ /**
+ *
+ */
struct Trail *succ_tail;
/**
};
+/**
+ *
+ */
struct FingerTable;
+/**
+ *
+ */
struct Finger
{
+ /**
+ *
+ */
struct Trail *trail;
+ /**
+ *
+ */
struct FingerTable *ft;
+ /**
+ *
+ */
struct GNUNET_HashCode destination;
/**
struct GNUNET_MessageHeader header;
/**
- * Zero, for alignment.
+ * #GNUNET_YES if the path should be recorded, #GNUNET_NO if not; in NBO.
*/
- uint32_t reserved GNUNET_PACKED;
+ uint16_t record_path GNUNET_PACKED;
+
+ /**
+ * Length of the recorded trail, 0 if @e record_path is #GNUNET_NO; in NBO.
+ */
+ uint16_t path_length GNUNET_PACKED;
/**
* Unique (random) identifier this peer will use to
*/
struct GNUNET_HashCode trail_id;
- /* followed by payload to send along the trail */
+ /**
+ * Path the message has taken so far (excluding sender).
+ */
+ /* struct GNUNET_PeerIdentity path[path_length]; */
+
+ /* followed by payload (another `struct GNUNET_MessageHeader`) to
+ send along the trail */
};
enum GNUNET_DHT_RouteOption options,
uint32_t desired_replication_level,
struct GNUNET_TIME_Absolute expiration_time,
- const void *data, size_t data_size)
+ const void *data,
+ size_t data_size)
{
GDS_DATACACHE_handle_put (expiration_time,
key,
0, NULL,
+ 0, NULL,
block_type,
data_size,
data);
+ GDS_CLIENTS_process_put (options,
+ block_type,
+ 0, 0,
+ 0, NULL,
+ expiration_time,
+ key,
+ data,
+ data_size);
}
struct TrailDestroyMessage *tdm;
struct Finger *finger;
- friend = current->pred;
+ friend = trail->pred;
if (NULL != friend)
{
if (GNUNET_YES == inform_pred)
{
env = GNUNET_MQ_msg (tdm,
GNUNET_MESSAGE_TYPE_WDHT_TRAIL_DESTROY);
- tdm->trail_id = current->pred_id;
+ tdm->trail_id = trail->pred_id;
GNUNET_MQ_send (friend->mq,
env);
}
friend->pred_tail,
trail);
}
- friend = current->succ;
+ friend = trail->succ;
if (NULL != friend)
{
if (GNUNET_YES == inform_succ)
{
env = GNUNET_MQ_msg (tdm,
GNUNET_MESSAGE_TYPE_WDHT_TRAIL_DESTROY);
- tdm->trail_id = current->pred_id;
+ tdm->trail_id = trail->pred_id;
GNUNET_MQ_send (friend->mq,
env);
}
}
+/**
+ * Blah.
+ */
+static void
+forward_message_on_trail (struct FriendInfo *next_target,
+ const struct GNUNET_HashCode *trail_id,
+ int have_path,
+ const struct GNUNET_PeerIdentity *predecessor,
+ const struct GNUNET_PeerIdentity *path,
+ uint16_t path_length,
+ const struct GNUNET_MessageHeader *payload)
+{
+ struct GNUNET_MQ_Envelope *env;
+ struct TrailRouteMessage *trm;
+ struct GNUNET_PeerIdentity *new_path;
+ unsigned int plen;
+ uint16_t payload_len;
+
+ payload_len = ntohs (payload->size);
+ if (have_path)
+ {
+ plen = path_length + 1;
+ if (plen >= (GNUNET_SERVER_MAX_MESSAGE_SIZE
+ - payload_len
+ - sizeof (struct TrailRouteMessage))
+ / sizeof (struct GNUNET_PeerIdentity))
+ {
+ /* Should really not have paths this long... */
+ GNUNET_break_op (0);
+ plen = 0;
+ have_path = 0;
+ }
+ }
+ else
+ {
+ GNUNET_break_op (0 == path_length);
+ path_length = 0;
+ plen = 0;
+ }
+ env = GNUNET_MQ_msg_extra (trm,
+ payload_len +
+ plen * sizeof (struct GNUNET_PeerIdentity),
+ GNUNET_MESSAGE_TYPE_WDHT_TRAIL_ROUTE);
+ trm->record_path = htons (have_path);
+ trm->path_length = htons (plen);
+ trm->trail_id = *trail_id;
+ new_path = (struct GNUNET_PeerIdentity *) &trm[1];
+ if (have_path)
+ {
+ memcpy (new_path,
+ path,
+ path_length * sizeof (struct GNUNET_PeerIdentity));
+ new_path[path_length] = *predecessor;
+ }
+ memcpy (&new_path[plen],
+ payload,
+ payload_len);
+ GNUNET_MQ_send (next_target->mq,
+ env);
+}
+
+
/**
* Send the get result to requesting client.
*
+ * @param trail_id trail identifying where to send the result to, NULL for us
* @param key Key of the requested data.
* @param type Block type
- * @param target_peer Next peer to forward the message to.
- * @param source_peer Peer which has the data for the key.
* @param put_path_length Number of peers in @a put_path
* @param put_path Path taken to put the data at its stored location.
- * @param get_path_length Number of peers in @a get_path
- * @param get_path Path taken to reach to the location of the key.
* @param expiration When will this result expire?
* @param data Payload to store
* @param data_size Size of the @a data
+ *
+ * FIXME: also pass options, so we know to record paths or not...
*/
void
-GDS_NEIGHBOURS_send_get_result (const struct GNUNET_HashCode *key,
+GDS_NEIGHBOURS_send_get_result (const struct GNUNET_HashCode *trail_id,
+ const struct GNUNET_HashCode *key,
enum GNUNET_BLOCK_Type type,
- const struct GNUNET_PeerIdentity *target_peer,
- const struct GNUNET_PeerIdentity *source_peer,
unsigned int put_path_length,
const struct GNUNET_PeerIdentity *put_path,
- unsigned int get_path_length,
- const struct GNUNET_PeerIdentity *get_path,
struct GNUNET_TIME_Absolute expiration,
- const void *data, size_t data_size)
+ const void *data,
+ size_t data_size)
{
- // TRICKY: need to introduce some context to remember trail from
- // the lookup...
+ struct GNUNET_MessageHeader *payload;
+
+ payload = GNUNET_malloc (sizeof (struct GNUNET_MessageHeader) + data_size);
+ payload->size = data_size;
+ payload->type = GNUNET_MESSAGE_TYPE_WDHT_GET_RESULT;
+
+ forward_message_on_trail (NULL /* FIXME: put something right */,
+ trail_id,
+ 0 /* FIXME: put something right */,
+ &my_identity,
+ put_path,
+ put_path_length,
+ payload);
}
+
/**
* Method called whenever a peer disconnects.
*
static struct FriendInfo *
pick_random_friend ()
{
- // TODO: need to extend peermap API to return random entry...
- // (Note: same extension exists for hashmap API).
- return NULL; // FIXME...
+ GNUNET_CONTAINER_PeerMapIterator *it;
+ if (0 != GNUNET_CONTAINER_multipeermap_get_random (friends_peermap,
+ *it,
+ NULL) ){
+ static struct FriendInfo **friend;
+ struct GNUNET_PeerIdentity *key;
+
+ /* FIXME: i am not sure of this one */
+ key = NULL;
+
+ if(GNUNET_YES == GNUNET_CONTAINER_multipeermap_iterator_next (it,
+ key,
+ (void *) friend))
+ {
+ return *friend;
+ }
+ }
+ return NULL;
+}
+
+
+/**
+ * One of our trails might have timed out, check and
+ * possibly initiate cleanup.
+ *
+ * @param cls NULL
+ * @param tc unused
+ */
+static void
+trail_timeout_callback (void *cls,
+ const struct GNUNET_SCHEDULER_TaskContext *tc)
+{
+ struct Trail *trail;
+ struct GNUNET_TIME_Relative left;
+
+ trail_timeout_task = NULL;
+ while (NULL != (trail = GNUNET_CONTAINER_heap_peek (trail_heap)))
+ {
+ left = GNUNET_TIME_absolute_get_remaining (trail->expiration_time);
+ if (0 != left.rel_value_us)
+ break;
+ delete_trail (trail,
+ GNUNET_YES,
+ GNUNET_YES);
+ }
+ if (NULL != trail)
+ trail_timeout_task = GNUNET_SCHEDULER_add_delayed (left,
+ &trail_timeout_callback,
+ NULL);
}
friend->succ_head,
friend->succ_tail,
trail);
+ trail->expiration_time = GNUNET_TIME_relative_to_absolute (TRAIL_TIMEOUT);
+ trail->hn = GNUNET_CONTAINER_heap_insert (trail_heap,
+ trail,
+ trail->expiration_time.abs_value_us);
+ if (NULL == trail_timeout_task)
+ trail_timeout_task = GNUNET_SCHEDULER_add_delayed (TRAIL_TIMEOUT,
+ &trail_timeout_callback,
+ NULL);
env = GNUNET_MQ_msg (rwm,
GNUNET_MESSAGE_TYPE_WDHT_RANDOM_WALK);
rwm->hops_taken = htonl (0);
if (ft->finger_array_size < 42)
{
// FIXME: must have finger array of the right size here,
- // FIXME: growing / shrinking are tricy -- with pointers
+ // FIXME: growing / shrinking are tricky -- with pointers
// from Trails!!!
}
t = GNUNET_new (struct Trail);
t->pred_id = m->trail_id;
t->pred = pred;
- t->expiration_time = GNUNET_TIME_relative_to_absolute (TRAIL_TIMEOUT);
if (GNUNET_OK !=
GNUNET_CONTAINER_multihashmap_put (trail_map,
&t->pred_id,
pred->pred_head,
pred->pred_tail,
t);
+ t->expiration_time = GNUNET_TIME_relative_to_absolute (TRAIL_TIMEOUT);
+ t->hn = GNUNET_CONTAINER_heap_insert (trail_heap,
+ t,
+ t->expiration_time.abs_value_us);
+ if (NULL == trail_timeout_task)
+ trail_timeout_task = GNUNET_SCHEDULER_add_delayed (TRAIL_TIMEOUT,
+ &trail_timeout_callback,
+ NULL);
+
if (ntohl (m->hops_taken) > GDS_NSE_get ())
{
/* We are the last hop, generate response */
const struct RandomWalkResponseMessage *rwrm;
rwrm = (const struct RandomWalkResponseMessage *) message;
- // 1) lookup trail => find Finger entry => fill in 'destination' and mark valid, move to end of sorted array, mark unsorted, update links from 'trails'
+ // 1) lookup trail => find Finger entry => fill in 'destination' and mark valid, move to end of sorted array,
+ //mark unsorted, update links from 'trails'
/*
* Steps :
* 1 check if we are the correct layer
const struct GNUNET_MessageHeader *message)
{
const struct TrailDestroyMessage *tdm;
+ struct Trail *trail;
tdm = (const struct TrailDestroyMessage *) message;
-
- /*
- * Steps :
- * 1 check if message comme from a trail (that we still remember...)
- * 1.a.1 if true: send the destroy message to the rest trail
- * 1.a.2 clean the trail structure
- * 1.a.3 did i have to remove the trail and ID from the db structure?
- * 1.b if false: do nothing
- */
-
- return GNUNET_OK;
-}
-
-
-/**
- * Handle a `struct TrailRouteMessage`.
- *
- * @param cls closure (NULL)
- * @param peer sender identity
- * @param message the finger destroy message
- * @return #GNUNET_OK on success, #GNUNET_SYSERR on error
- */
-static int
-handle_dht_p2p_trail_route (void *cls,
- const struct GNUNET_PeerIdentity *peer,
- const struct GNUNET_MessageHeader *message)
-{
- const struct TrailRouteMessage *trm;
-
- trm = (const struct TrailRouteMessage *) message;
-
- /*
- * Steps :
- * 1 check if message comme from a trail
- * 1.a.1 if trail not finished with us, continue to forward
- * 1.a.2 otherwise handle body message embedded in trail
- */
-
+ trail = GNUNET_CONTAINER_multihashmap_get (trail_map,
+ &tdm->trail_id);
+ delete_trail (trail,
+ ( (NULL != trail->succ) &&
+ (0 == memcmp (peer,
+ &trail->succ->id,
+ sizeof (struct GNUNET_PeerIdentity))) ),
+ ( (NULL != trail->pred) &&
+ (0 == memcmp (peer,
+ &trail->pred->id,
+ sizeof (struct GNUNET_PeerIdentity))) ));
return GNUNET_OK;
}
* message.
*
* @param cls closure (NULL)
- * @param peer sender identity
+ * @param trail_id path to the originator
* @param message the finger setup message
* @return #GNUNET_OK on success, #GNUNET_SYSERR on error
*/
static int
handle_dht_p2p_successor_find (void *cls,
- const struct GNUNET_PeerIdentity *peer,
+ const struct GNUNET_HashCode *trail_id,
const struct GNUNET_MessageHeader *message)
{
const struct FindSuccessorMessage *fsm;
// locate trail (for sending reply), if not exists, fail nicely.
// otherwise, go to datacache and return 'top k' elements closest to 'key'
// as "PUT" messages via the trail (need to extend DB API!)
-
+#if 0
+ GDS_DATACACHE_get_successors (trail_id,
+ key);
+#endif
return GNUNET_OK;
}
* Handle a `struct PeerGetMessage`.
*
* @param cls closure (NULL)
- * @param peer sender identity
+ * @param trail_id path to the originator
* @param message the peer get message
* @return #GNUNET_OK on success, #GNUNET_SYSERR on error
*/
static int
handle_dht_p2p_peer_get (void *cls,
- const struct GNUNET_PeerIdentity *peer,
+ const struct GNUNET_HashCode *trail_id,
const struct GNUNET_MessageHeader *message)
{
const struct PeerGetMessage *pgm;
* Handle a `struct PeerGetResultMessage`.
*
* @param cls closure (NULL)
- * @param peer sender identity
+ * @param trail_id path to the originator
* @param message the peer get result message
* @return #GNUNET_OK on success, #GNUNET_SYSERR on error
*/
static int
handle_dht_p2p_peer_get_result (void *cls,
- const struct GNUNET_PeerIdentity *peer,
- const struct GNUNET_MessageHeader *message)
+ const struct GNUNET_HashCode *trail_id,
+ const struct GNUNET_MessageHeader *message)
{
const struct PeerGetResultMessage *pgrm;
pgrm = (const struct PeerGetResultMessage *) message;
// pretty much: parse, & pass to client (there is some call for that...)
+#if 0
+ GDS_CLIENTS_process_get (options,
+ type,
+ 0, 0,
+ path_length, path,
+ key);
+ (void) GDS_DATACACHE_handle_get (trail_id,
+ key,
+ type,
+ xquery,
+ xquery_size,
+ &reply_bf,
+ reply_bf_mutator);
+#endif
return GNUNET_OK;
}
* Handle a `struct PeerPutMessage`.
*
* @param cls closure (NULL)
- * @param peer sender identity
+ * @param trail_id path to the originator
* @param message the peer put message
* @return #GNUNET_OK on success, #GNUNET_SYSERR on error
*/
static int
handle_dht_p2p_peer_put (void *cls,
- const struct GNUNET_PeerIdentity *peer,
- const struct GNUNET_MessageHeader *message)
+ const struct GNUNET_HashCode *trail_id,
+ const struct GNUNET_MessageHeader *message)
{
const struct PeerGetResultMessage *pgrm;
* 2 use the API to add the value in the "database". Check on the xdht file, how to do it.
* 3 Did i a have to return a notification or did i have to return GNUNET_[OK|SYSERR]?
*/
+#if 0
+ GDS_DATACACHE_handle_put (expiration_time,
+ key,
+ path_length, path,
+ block_type,
+ data_size,
+ data);
+ GDS_CLIENTS_process_put (options,
+ block_type,
+ 0, 0,
+ path_length, path,
+ expiration_time,
+ key,
+ data,
+ data_size);
+#endif
+ return GNUNET_OK;
+}
+
+
+
+
+/**
+ * Handler for a message we received along some trail.
+ *
+ * @param cls closure
+ * @param trail_id trail identifier
+ * @param message the message we got
+ * @return #GNUNET_OK on success, #GNUNET_SYSERR on error
+ */
+typedef int
+(*TrailHandlerCallback)(void *cls,
+ const struct GNUNET_HashCode *trail_id,
+ const struct GNUNET_MessageHeader *message);
+
+
+/**
+ * Definition of a handler for a message received along some trail.
+ */
+struct TrailHandler
+{
+ /**
+ * NULL for end-of-list.
+ */
+ TrailHandlerCallback callback;
+
+ /**
+ * Closure for @e callback.
+ */
+ void *cls;
+
+ /**
+ * Message type this handler addresses.
+ */
+ uint16_t message_type;
+
+ /**
+ * Use 0 for variable-size.
+ */
+ uint16_t message_size;
+};
+
+
+/**
+ * Handle a `struct TrailRouteMessage`.
+ *
+ * @param cls closure (NULL)
+ * @param peer sender identity
+ * @param message the finger destroy message
+ * @return #GNUNET_OK on success, #GNUNET_SYSERR on error
+ */
+static int
+handle_dht_p2p_trail_route (void *cls,
+ const struct GNUNET_PeerIdentity *peer,
+ const struct GNUNET_MessageHeader *message)
+{
+ static const struct TrailHandler handlers[] = {
+ { &handle_dht_p2p_successor_find, NULL,
+ GNUNET_MESSAGE_TYPE_WDHT_SUCCESSOR_FIND,
+ sizeof (struct FindSuccessorMessage) },
+ { NULL, NULL, 0, 0 }
+ };
+ unsigned int i;
+ const struct TrailRouteMessage *trm;
+ const struct GNUNET_PeerIdentity *path;
+ uint16_t path_length;
+ const struct GNUNET_MessageHeader *payload;
+ const struct TrailHandler *th;
+ struct Trail *trail;
+ size_t msize;
+
+ /* Parse and check message is well-formed */
+ msize = ntohs (message->size);
+ if (msize < sizeof (struct TrailRouteMessage))
+ {
+ GNUNET_break_op (0);
+ return GNUNET_YES;
+ }
+ trm = (const struct TrailRouteMessage *) message;
+ path_length = ntohs (trm->path_length);
+ if (msize < sizeof (struct TrailRouteMessage) +
+ path_length * sizeof (struct GNUNET_PeerIdentity) +
+ sizeof (struct GNUNET_MessageHeader) )
+ {
+ GNUNET_break_op (0);
+ return GNUNET_YES;
+ }
+ path = (const struct GNUNET_PeerIdentity *) &trm[1];
+ payload = (const struct GNUNET_MessageHeader *) &path[path_length];
+ if (msize != (ntohs (payload->size) +
+ sizeof (struct TrailRouteMessage) +
+ path_length * sizeof (struct GNUNET_PeerIdentity)))
+ {
+ GNUNET_break_op (0);
+ return GNUNET_YES;
+ }
+
+ /* Is this message for us? */
+ trail = GNUNET_CONTAINER_multihashmap_get (trail_map,
+ &trm->trail_id);
+ if ( (NULL != trail->pred) &&
+ (0 == memcmp (peer,
+ &trail->pred->id,
+ sizeof (struct GNUNET_PeerIdentity))) )
+ {
+ /* forward to 'successor' */
+ if (NULL != trail->succ)
+ {
+ forward_message_on_trail (trail->succ,
+ &trail->succ_id,
+ ntohs (trm->record_path),
+ peer,
+ path,
+ path_length,
+ payload);
+ return GNUNET_OK;
+ }
+ }
+ else
+ {
+ /* forward to 'predecessor' */
+ GNUNET_break_op ( (NULL != trail->succ) &&
+ (0 == memcmp (peer,
+ &trail->succ->id,
+ sizeof (struct GNUNET_PeerIdentity))) );
+ if (NULL != trail->pred)
+ {
+ forward_message_on_trail (trail->pred,
+ &trail->pred_id,
+ ntohs (trm->record_path),
+ peer,
+ path,
+ path_length,
+ payload);
+ return GNUNET_OK;
+ }
+ }
+
+ /* Message is for us, dispatch to handler */
+ th = NULL;
+ for (i=0; NULL != handlers[i].callback; i++)
+ {
+ th = &handlers[i];
+ if (ntohs (payload->type) == th->message_type)
+ {
+ if ( (0 == th->message_size) ||
+ (ntohs (payload->size) == th->message_size) )
+ th->callback (th->cls,
+ &trm->trail_id,
+ payload);
+ else
+ GNUNET_break_op (0);
+ break;
+ }
+ }
+ GNUNET_break_op (NULL != th);
return GNUNET_OK;
}
{ &handle_dht_p2p_trail_route,
GNUNET_MESSAGE_TYPE_WDHT_TRAIL_ROUTE,
0},
- { &handle_dht_p2p_successor_find,
- GNUNET_MESSAGE_TYPE_WDHT_SUCCESSOR_FIND,
- sizeof (struct FindSuccessorMessage) },
- { &handle_dht_p2p_peer_get,
- GNUNET_MESSAGE_TYPE_WDHT_GET,
- sizeof (struct PeerGetMessage) },
- { &handle_dht_p2p_peer_get_result,
- GNUNET_MESSAGE_TYPE_WDHT_GET_RESULT,
- 0},
- { &handle_dht_p2p_peer_put,
- GNUNET_MESSAGE_TYPE_WDHT_PUT,
- 0},
{NULL, 0, 0}
};
trail_map = NULL;
GNUNET_CONTAINER_heap_destroy (trail_heap);
trail_heap = NULL;
+ if (NULL != trail_timeout_task)
+ {
+ GNUNET_SCHEDULER_cancel (trail_timeout_task);
+ trail_timeout_task = NULL;
+ }
}
projects / oweals / gnunet.git / blobdiff