[Top][All Lists]
[Date Prev][Date Next][Thread Prev][Thread Next][Date Index][Thread Index]
[GNUnet-SVN] r16976 - gnunet/src/dht
|
From: |
gnunet |
|
Subject: |
[GNUnet-SVN] r16976 - gnunet/src/dht |
|
Date: |
Wed, 21 Sep 2011 07:51:38 +0200 |
Author: grothoff
Date: 2011-09-21 07:51:38 +0200 (Wed, 21 Sep 2011)
New Revision: 16976
Added:
gnunet/src/dht/dht_api_new.c
gnunet/src/dht/dht_new.h
gnunet/src/dht/gnunet-service-dht-new.c
gnunet/src/dht/gnunet-service-dht_clients.c
gnunet/src/dht/gnunet-service-dht_clients.h
gnunet/src/dht/gnunet-service-dht_neighbours.c
gnunet/src/dht/gnunet-service-dht_neighbours.h
gnunet/src/dht/gnunet-service-dht_nse.c
gnunet/src/dht/gnunet-service-dht_nse.h
Log:
add
Added: gnunet/src/dht/dht_api_new.c
===================================================================
--- gnunet/src/dht/dht_api_new.c (rev 0)
+++ gnunet/src/dht/dht_api_new.c 2011-09-21 05:51:38 UTC (rev 16976)
@@ -0,0 +1,909 @@
+/*
+ This file is part of GNUnet.
+ (C) 2009, 2010 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
+ 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
+ WITHOUT ANY WARRANTY; without even the implied warranty of
+ MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ General Public License for more details.
+
+ 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.
+*/
+
+/**
+ * @file dht/dht_api.c
+ * @brief library to access the DHT service
+ * @author Christian Grothoff
+ * @author Nathan Evans
+ */
+
+#include "platform.h"
+#include "gnunet_bandwidth_lib.h"
+#include "gnunet_client_lib.h"
+#include "gnunet_constants.h"
+#include "gnunet_container_lib.h"
+#include "gnunet_arm_service.h"
+#include "gnunet_hello_lib.h"
+#include "gnunet_protocols.h"
+#include "gnunet_server_lib.h"
+#include "gnunet_time_lib.h"
+#include "gnunet_dht_service.h"
+#include "dht.h"
+
+#define DEBUG_DHT_API GNUNET_NO
+
+/**
+ * Entry in our list of messages to be (re-)transmitted.
+ */
+struct PendingMessage
+{
+ /**
+ * This is a doubly-linked list.
+ */
+ struct PendingMessage *prev;
+
+ /**
+ * This is a doubly-linked list.
+ */
+ struct PendingMessage *next;
+
+ /**
+ * Message that is pending, allocated at the end
+ * of this struct.
+ */
+ const struct GNUNET_MessageHeader *msg;
+
+ /**
+ * Handle to the DHT API context.
+ */
+ struct GNUNET_DHT_Handle *handle;
+
+ /**
+ * Continuation to call when the request has been
+ * transmitted (for the first time) to the service; can be NULL.
+ */
+ GNUNET_SCHEDULER_Task cont;
+
+ /**
+ * Closure for 'cont'.
+ */
+ void *cont_cls;
+
+ /**
+ * Timeout task for this message
+ */
+ GNUNET_SCHEDULER_TaskIdentifier timeout_task;
+
+ /**
+ * Unique ID for this request
+ */
+ uint64_t unique_id;
+
+ /**
+ * Free the saved message once sent, set to GNUNET_YES for messages
+ * that do not receive responses; GNUNET_NO if this pending message
+ * is aliased from a 'struct GNUNET_DHT_RouteHandle' and will be freed
+ * from there.
+ */
+ int free_on_send;
+
+ /**
+ * GNUNET_YES if this message is in our pending queue right now.
+ */
+ int in_pending_queue;
+
+};
+
+
+/**
+ * Handle to a route request
+ */
+struct GNUNET_DHT_RouteHandle
+{
+
+ /**
+ * Iterator to call on data receipt
+ */
+ GNUNET_DHT_ReplyProcessor iter;
+
+ /**
+ * Closure for the iterator callback
+ */
+ void *iter_cls;
+
+ /**
+ * Main handle to this DHT api
+ */
+ struct GNUNET_DHT_Handle *dht_handle;
+
+ /**
+ * The actual message sent for this request,
+ * used for retransmitting requests on service
+ * failure/reconnect. Freed on route_stop.
+ */
+ struct PendingMessage *message;
+
+ /**
+ * Key that this get request is for
+ */
+ GNUNET_HashCode key;
+
+ /**
+ * Unique identifier for this request (for key collisions). FIXME:
redundant!?
+ */
+ uint64_t uid;
+
+};
+
+
+/**
+ * Connection to the DHT service.
+ */
+struct GNUNET_DHT_Handle
+{
+
+ /**
+ * Configuration to use.
+ */
+ const struct GNUNET_CONFIGURATION_Handle *cfg;
+
+ /**
+ * Socket (if available).
+ */
+ struct GNUNET_CLIENT_Connection *client;
+
+ /**
+ * Currently pending transmission request (or NULL).
+ */
+ struct GNUNET_CLIENT_TransmitHandle *th;
+
+ /**
+ * Head of linked list of messages we would like to transmit.
+ */
+ struct PendingMessage *pending_head;
+
+ /**
+ * Tail of linked list of messages we would like to transmit.
+ */
+ struct PendingMessage *pending_tail;
+
+ /**
+ * Hash map containing the current outstanding unique requests
+ * (values are of type 'struct GNUNET_DHT_RouteHandle').
+ */
+ struct GNUNET_CONTAINER_MultiHashMap *active_requests;
+
+ /**
+ * Task for trying to reconnect.
+ */
+ GNUNET_SCHEDULER_TaskIdentifier reconnect_task;
+
+ /**
+ * How quickly should we retry? Used for exponential back-off on
+ * connect-errors.
+ */
+ struct GNUNET_TIME_Relative retry_time;
+
+ /**
+ * Generator for unique ids.
+ */
+ uint64_t uid_gen;
+
+};
+
+
+/**
+ * Transmit the next pending message, called by notify_transmit_ready
+ */
+static size_t
+transmit_pending (void *cls, size_t size, void *buf);
+
+
+/**
+ * Handler for messages received from the DHT service
+ * a demultiplexer which handles numerous message types
+ *
+ */
+static void
+service_message_handler (void *cls, const struct GNUNET_MessageHeader *msg);
+
+
+
+
+/**
+ * Try to (re)connect to the DHT service.
+ *
+ * @return GNUNET_YES on success, GNUNET_NO on failure.
+ */
+static int
+try_connect (struct GNUNET_DHT_Handle *handle)
+{
+ if (handle->client != NULL)
+ return GNUNET_OK;
+ handle->client = GNUNET_CLIENT_connect ("dht", handle->cfg);
+ if (handle->client == NULL)
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+ _("Failed to connect to the DHT service!\n"));
+ return GNUNET_NO;
+ }
+#if DEBUG_DHT
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Starting to process replies from DHT\n");
+#endif
+ GNUNET_CLIENT_receive (handle->client, &service_message_handler, handle,
+ GNUNET_TIME_UNIT_FOREVER_REL);
+ return GNUNET_YES;
+}
+
+
+/**
+ * Add the request corresponding to the given route handle
+ * to the pending queue (if it is not already in there).
+ *
+ * @param cls the 'struct GNUNET_DHT_Handle*'
+ * @param key key for the request (not used)
+ * @param value the 'struct GNUNET_DHT_RouteHandle*'
+ * @return GNUNET_YES (always)
+ */
+static int
+add_request_to_pending (void *cls, const GNUNET_HashCode * key, void *value)
+{
+ struct GNUNET_DHT_Handle *handle = cls;
+ struct GNUNET_DHT_RouteHandle *rh = value;
+
+ if (GNUNET_NO == rh->message->in_pending_queue)
+ {
+ GNUNET_CONTAINER_DLL_insert (handle->pending_head, handle->pending_tail,
+ rh->message);
+ rh->message->in_pending_queue = GNUNET_YES;
+ }
+ return GNUNET_YES;
+}
+
+
+/**
+ * Try to send messages from list of messages to send
+ * @param handle DHT_Handle
+ */
+static void
+process_pending_messages (struct GNUNET_DHT_Handle *handle);
+
+
+/**
+ * Try reconnecting to the dht service.
+ *
+ * @param cls GNUNET_DHT_Handle
+ * @param tc scheduler context
+ */
+static void
+try_reconnect (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
+{
+ struct GNUNET_DHT_Handle *handle = cls;
+
+ handle->reconnect_task = GNUNET_SCHEDULER_NO_TASK;
+ if (handle->retry_time.rel_value < GNUNET_CONSTANTS_SERVICE_RETRY.rel_value)
+ handle->retry_time = GNUNET_CONSTANTS_SERVICE_RETRY;
+ else
+ handle->retry_time = GNUNET_TIME_relative_multiply (handle->retry_time, 2);
+ if (handle->retry_time.rel_value >
GNUNET_CONSTANTS_SERVICE_TIMEOUT.rel_value)
+ handle->retry_time = GNUNET_CONSTANTS_SERVICE_TIMEOUT;
+ handle->reconnect_task = GNUNET_SCHEDULER_NO_TASK;
+ handle->client = GNUNET_CLIENT_connect ("dht", handle->cfg);
+ if (handle->client == NULL)
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "dht reconnect failed(!)\n");
+ return;
+ }
+ GNUNET_CONTAINER_multihashmap_iterate (handle->active_requests,
+ &add_request_to_pending, handle);
+ process_pending_messages (handle);
+}
+
+
+/**
+ * Try reconnecting to the DHT service.
+ *
+ * @param handle handle to dht to (possibly) disconnect and reconnect
+ */
+static void
+do_disconnect (struct GNUNET_DHT_Handle *handle)
+{
+ if (handle->client == NULL)
+ return;
+ GNUNET_assert (handle->reconnect_task == GNUNET_SCHEDULER_NO_TASK);
+ if (NULL != handle->th)
+ GNUNET_CLIENT_notify_transmit_ready_cancel(handle->th);
+ handle->th = NULL;
+ GNUNET_CLIENT_disconnect (handle->client, GNUNET_NO);
+ handle->client = NULL;
+ handle->reconnect_task =
+ GNUNET_SCHEDULER_add_delayed (handle->retry_time, &try_reconnect,
handle);
+}
+
+
+/**
+ * Try to send messages from list of messages to send
+ */
+static void
+process_pending_messages (struct GNUNET_DHT_Handle *handle)
+{
+ struct PendingMessage *head;
+
+ if (handle->client == NULL)
+ {
+ do_disconnect (handle);
+ return;
+ }
+ if (handle->th != NULL)
+ return;
+ if (NULL == (head = handle->pending_head))
+ return;
+ handle->th =
+ GNUNET_CLIENT_notify_transmit_ready (handle->client,
+ ntohs (head->msg->size),
+ GNUNET_TIME_UNIT_FOREVER_REL,
+ GNUNET_YES, &transmit_pending,
+ handle);
+ if (NULL == handle->th)
+ {
+ do_disconnect (handle);
+ return;
+ }
+}
+
+
+/**
+ * Transmit the next pending message, called by notify_transmit_ready
+ */
+static size_t
+transmit_pending (void *cls, size_t size, void *buf)
+{
+ struct GNUNET_DHT_Handle *handle = cls;
+ struct PendingMessage *head;
+ size_t tsize;
+
+ handle->th = NULL;
+ if (buf == NULL)
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Transmission to DHT service failed! Reconnecting!\n");
+ do_disconnect (handle);
+ return 0;
+ }
+ if (NULL == (head = handle->pending_head))
+ return 0;
+
+ tsize = ntohs (head->msg->size);
+ if (size < tsize)
+ {
+ process_pending_messages (handle);
+ return 0;
+ }
+ memcpy (buf, head->msg, tsize);
+ GNUNET_CONTAINER_DLL_remove (handle->pending_head, handle->pending_tail,
+ head);
+ if (head->timeout_task != GNUNET_SCHEDULER_NO_TASK)
+ {
+ GNUNET_SCHEDULER_cancel (head->timeout_task);
+ head->timeout_task = GNUNET_SCHEDULER_NO_TASK;
+ }
+ if (NULL != head->cont)
+ {
+ GNUNET_SCHEDULER_add_continuation (head->cont, head->cont_cls,
+ GNUNET_SCHEDULER_REASON_PREREQ_DONE);
+ head->cont = NULL;
+ head->cont_cls = NULL;
+ }
+ head->in_pending_queue = GNUNET_NO;
+ if (GNUNET_YES == head->free_on_send)
+ GNUNET_free (head);
+ process_pending_messages (handle);
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Forwarded request of %u bytes to DHT service\n",
+ (unsigned int) tsize);
+ return tsize;
+}
+
+
+/**
+ * Process a given reply that might match the given
+ * request.
+ */
+static int
+process_reply (void *cls, const GNUNET_HashCode * key, void *value)
+{
+ const struct GNUNET_DHT_RouteResultMessage *dht_msg = cls;
+ struct GNUNET_DHT_RouteHandle *rh = value;
+ const struct GNUNET_MessageHeader *enc_msg;
+ size_t enc_size;
+ uint64_t uid;
+ const struct GNUNET_PeerIdentity **outgoing_path;
+ const struct GNUNET_PeerIdentity *pos;
+ uint32_t outgoing_path_length;
+ unsigned int i;
+ char *path_offset;
+
+ uid = GNUNET_ntohll (dht_msg->unique_id);
+#if HAVE_UID_FOR_TESTING
+ if (uid != rh->uid)
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Reply UID did not match request UID\n");
+ return GNUNET_YES;
+ }
+#endif
+ enc_msg = (const struct GNUNET_MessageHeader *) &dht_msg[1];
+ enc_size = ntohs (enc_msg->size);
+ if (enc_size < sizeof (struct GNUNET_MessageHeader))
+ {
+ GNUNET_break (0);
+ return GNUNET_NO;
+ }
+ path_offset = (char *) &dht_msg[1];
+ path_offset += enc_size;
+ pos = (const struct GNUNET_PeerIdentity *) path_offset;
+ outgoing_path_length = ntohl (dht_msg->outgoing_path_length);
+ if (outgoing_path_length * sizeof (struct GNUNET_PeerIdentity) >
+ ntohs (dht_msg->header.size) - enc_size)
+ {
+ GNUNET_break (0);
+ return GNUNET_NO;
+ }
+
+ if (outgoing_path_length > 0)
+ {
+ outgoing_path =
+ GNUNET_malloc ((outgoing_path_length +
+ 1) * sizeof (struct GNUNET_PeerIdentity *));
+ for (i = 0; i < outgoing_path_length; i++)
+ {
+ outgoing_path[i] = pos;
+ pos++;
+ }
+ outgoing_path[outgoing_path_length] = NULL;
+ }
+ else
+ outgoing_path = NULL;
+
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Processing reply.\n");
+ rh->iter (rh->iter_cls, &rh->key, outgoing_path, enc_msg);
+ GNUNET_free_non_null (outgoing_path);
+ return GNUNET_YES;
+}
+
+
+/**
+ * Handler for messages received from the DHT service
+ * a demultiplexer which handles numerous message types
+ *
+ * @param cls the 'struct GNUNET_DHT_Handle'
+ * @param msg the incoming message
+ */
+static void
+service_message_handler (void *cls, const struct GNUNET_MessageHeader *msg)
+{
+ struct GNUNET_DHT_Handle *handle = cls;
+ const struct GNUNET_DHT_RouteResultMessage *dht_msg;
+
+ if (msg == NULL)
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Error receiving data from DHT service, reconnecting\n");
+ do_disconnect (handle);
+ return;
+ }
+ if (ntohs (msg->type) != GNUNET_MESSAGE_TYPE_DHT_LOCAL_ROUTE_RESULT)
+ {
+ GNUNET_break (0);
+ do_disconnect (handle);
+ return;
+ }
+ if (ntohs (msg->size) < sizeof (struct GNUNET_DHT_RouteResultMessage))
+ {
+ GNUNET_break (0);
+ do_disconnect (handle);
+ return;
+ }
+ dht_msg = (const struct GNUNET_DHT_RouteResultMessage *) msg;
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Comparing reply `%s' against %u pending requests.\n",
+ GNUNET_h2s (&dht_msg->key),
+ GNUNET_CONTAINER_multihashmap_size (handle->active_requests));
+ GNUNET_CONTAINER_multihashmap_get_multiple (handle->active_requests,
+ &dht_msg->key, &process_reply,
+ (void *) dht_msg);
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Continuing to process replies from DHT\n");
+ GNUNET_CLIENT_receive (handle->client, &service_message_handler, handle,
+ GNUNET_TIME_UNIT_FOREVER_REL);
+
+}
+
+
+/**
+ * Initialize the connection with the DHT service.
+ *
+ * @param cfg configuration to use
+ * @param ht_len size of the internal hash table to use for
+ * processing multiple GET/FIND requests in parallel
+ *
+ * @return handle to the DHT service, or NULL on error
+ */
+struct GNUNET_DHT_Handle *
+GNUNET_DHT_connect (const struct GNUNET_CONFIGURATION_Handle *cfg,
+ unsigned int ht_len)
+{
+ struct GNUNET_DHT_Handle *handle;
+
+ handle = GNUNET_malloc (sizeof (struct GNUNET_DHT_Handle));
+ handle->cfg = cfg;
+ handle->uid_gen =
+ GNUNET_CRYPTO_random_u64 (GNUNET_CRYPTO_QUALITY_WEAK, UINT64_MAX);
+ handle->active_requests = GNUNET_CONTAINER_multihashmap_create (ht_len);
+ if (GNUNET_NO == try_connect (handle))
+ {
+ GNUNET_DHT_disconnect (handle);
+ return NULL;
+ }
+ return handle;
+}
+
+
+/**
+ * Shutdown connection with the DHT service.
+ *
+ * @param handle handle of the DHT connection to stop
+ */
+void
+GNUNET_DHT_disconnect (struct GNUNET_DHT_Handle *handle)
+{
+ struct PendingMessage *pm;
+
+ GNUNET_assert (handle != NULL);
+ GNUNET_assert (0 ==
+ GNUNET_CONTAINER_multihashmap_size (handle->active_requests));
+ if (handle->th != NULL)
+ {
+ GNUNET_CLIENT_notify_transmit_ready_cancel (handle->th);
+ handle->th = NULL;
+ }
+ while (NULL != (pm = handle->pending_head))
+ {
+ GNUNET_CONTAINER_DLL_remove (handle->pending_head, handle->pending_tail,
+ pm);
+ GNUNET_assert (GNUNET_YES == pm->free_on_send);
+ if (GNUNET_SCHEDULER_NO_TASK != pm->timeout_task)
+ GNUNET_SCHEDULER_cancel (pm->timeout_task);
+ if (NULL != pm->cont)
+ GNUNET_SCHEDULER_add_continuation (pm->cont, pm->cont_cls,
+ GNUNET_SCHEDULER_REASON_TIMEOUT);
+ pm->in_pending_queue = GNUNET_NO;
+ GNUNET_free (pm);
+ }
+ if (handle->client != NULL)
+ {
+ GNUNET_CLIENT_disconnect (handle->client, GNUNET_YES);
+ handle->client = NULL;
+ }
+ if (handle->reconnect_task != GNUNET_SCHEDULER_NO_TASK)
+ GNUNET_SCHEDULER_cancel (handle->reconnect_task);
+ GNUNET_CONTAINER_multihashmap_destroy (handle->active_requests);
+ GNUNET_free (handle);
+}
+
+
+
+
+/* ***** Special low-level API providing generic routing abstraction ***** */
+
+
+/**
+ * Timeout for the transmission of a fire&forget-request. Clean it up.
+ *
+ * @param cls the 'struct PendingMessage'
+ * @param tc scheduler context
+ */
+static void
+timeout_route_request (void *cls, const struct GNUNET_SCHEDULER_TaskContext
*tc)
+{
+ struct PendingMessage *pending = cls;
+ struct GNUNET_DHT_Handle *handle;
+
+ if (pending->free_on_send != GNUNET_YES)
+ {
+ /* timeouts should only apply to fire & forget requests! */
+ GNUNET_break (0);
+ return;
+ }
+ handle = pending->handle;
+ GNUNET_CONTAINER_DLL_remove (handle->pending_head, handle->pending_tail,
+ pending);
+ if (pending->cont != NULL)
+ pending->cont (pending->cont_cls, tc);
+ GNUNET_free (pending);
+}
+
+
+/**
+ * Initiate a generic DHT route operation.
+ *
+ * @param handle handle to the DHT service
+ * @param key the key to look up
+ * @param desired_replication_level how many peers should ultimately receive
+ * this message (advisory only, target may be too high for the
+ * given DHT or not hit exactly).
+ * @param options options for routing
+ * @param enc send the encapsulated message to a peer close to the key
+ * @param iter function to call on each result, NULL if no replies are expected
+ * @param iter_cls closure for iter
+ * @param timeout when to abort with an error if we fail to get
+ * a confirmation for the request (when necessary) or how long
+ * to wait for tramission to the service; only applies
+ * if 'iter' is NULL
+ * @param cont continuation to call when the request has been transmitted
+ * the first time to the service
+ * @param cont_cls closure for cont
+ * @return handle to stop the request, NULL if the request is "fire and forget"
+ */
+struct GNUNET_DHT_RouteHandle *
+GNUNET_DHT_route_start (struct GNUNET_DHT_Handle *handle,
+ const GNUNET_HashCode * key,
+ uint32_t desired_replication_level,
+ enum GNUNET_DHT_RouteOption options,
+ const struct GNUNET_MessageHeader *enc,
+ struct GNUNET_TIME_Relative timeout,
+ GNUNET_DHT_ReplyProcessor iter, void *iter_cls,
+ GNUNET_SCHEDULER_Task cont, void *cont_cls)
+{
+ struct PendingMessage *pending;
+ struct GNUNET_DHT_RouteMessage *message;
+ struct GNUNET_DHT_RouteHandle *route_handle;
+ uint16_t msize;
+ uint16_t esize;
+
+ esize = ntohs (enc->size);
+ if (sizeof (struct GNUNET_DHT_RouteMessage) + esize >=
+ GNUNET_SERVER_MAX_MESSAGE_SIZE)
+ {
+ GNUNET_break (0);
+ return NULL;
+ }
+ msize = sizeof (struct GNUNET_DHT_RouteMessage) + esize;
+ pending = GNUNET_malloc (sizeof (struct PendingMessage) + msize);
+ message = (struct GNUNET_DHT_RouteMessage *) &pending[1];
+ pending->msg = &message->header;
+ pending->handle = handle;
+ pending->cont = cont;
+ pending->cont_cls = cont_cls;
+
+ message->header.size = htons (msize);
+ message->header.type = htons (GNUNET_MESSAGE_TYPE_DHT_LOCAL_ROUTE);
+ message->options = htonl ((uint32_t) options);
+ message->desired_replication_level = htonl (desired_replication_level);
+ message->reserved = 0;
+ message->key = *key;
+ handle->uid_gen++;
+ message->unique_id = GNUNET_htonll (handle->uid_gen);
+ memcpy (&message[1], enc, esize);
+
+ if (iter != NULL)
+ {
+ route_handle = GNUNET_malloc (sizeof (struct GNUNET_DHT_RouteHandle));
+ route_handle->key = *key;
+ route_handle->iter = iter;
+ route_handle->iter_cls = iter_cls;
+ route_handle->dht_handle = handle;
+ route_handle->uid = handle->uid_gen;
+ route_handle->message = pending;
+ GNUNET_CONTAINER_multihashmap_put (handle->active_requests, key,
+ route_handle,
+
GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
+ }
+ else
+ {
+ route_handle = NULL;
+ pending->free_on_send = GNUNET_YES;
+ pending->timeout_task =
+ GNUNET_SCHEDULER_add_delayed (timeout, &timeout_route_request,
pending);
+ }
+ GNUNET_CONTAINER_DLL_insert (handle->pending_head, handle->pending_tail,
+ pending);
+ pending->in_pending_queue = GNUNET_YES;
+ process_pending_messages (handle);
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "DHT route start request processed, returning %p\n",
+ route_handle);
+ return route_handle;
+}
+
+
+/**
+ * Stop a previously issued routing request
+ *
+ * @param route_handle handle to the request to stop
+ */
+void
+GNUNET_DHT_route_stop (struct GNUNET_DHT_RouteHandle *route_handle)
+{
+ struct GNUNET_DHT_Handle *handle;
+ struct PendingMessage *pending;
+ struct GNUNET_DHT_StopMessage *message;
+ size_t msize;
+
+ handle = route_handle->dht_handle;
+ if (GNUNET_NO == route_handle->message->in_pending_queue)
+ {
+ /* need to send stop message */
+ msize = sizeof (struct GNUNET_DHT_StopMessage);
+ pending = GNUNET_malloc (sizeof (struct PendingMessage) + msize);
+ message = (struct GNUNET_DHT_StopMessage *) &pending[1];
+ pending->msg = &message->header;
+ message->header.size = htons (msize);
+ message->header.type = htons (GNUNET_MESSAGE_TYPE_DHT_LOCAL_ROUTE_STOP);
+ message->reserved = 0;
+ message->unique_id = GNUNET_htonll (route_handle->uid);
+ message->key = route_handle->key;
+ pending->handle = handle;
+ pending->free_on_send = GNUNET_YES;
+ pending->in_pending_queue = GNUNET_YES;
+ GNUNET_CONTAINER_DLL_insert (handle->pending_head, handle->pending_tail,
+ pending);
+ process_pending_messages (handle);
+ }
+ else
+ {
+ /* simply remove pending request from message queue before
+ * transmission, no need to transmit STOP request! */
+ GNUNET_CONTAINER_DLL_remove (handle->pending_head, handle->pending_tail,
+ route_handle->message);
+ }
+ GNUNET_assert (GNUNET_YES ==
+ GNUNET_CONTAINER_multihashmap_remove
+ (route_handle->dht_handle->active_requests,
&route_handle->key,
+ route_handle));
+ GNUNET_free (route_handle->message);
+ GNUNET_free (route_handle);
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "DHT route stop request processed\n");
+}
+
+
+
+/* ***** Special API for controlling DHT routing maintenance ******* */
+
+
+/**
+ * Send a control message to the DHT.
+ *
+ * @param handle handle to the DHT service
+ * @param command command
+ * @param variable variable to the command
+ * @param cont continuation to call when done (transmitting request to service)
+ * @param cont_cls closure for cont
+ */
+static void
+send_control_message (struct GNUNET_DHT_Handle *handle, uint16_t command,
+ uint16_t variable, GNUNET_SCHEDULER_Task cont,
+ void *cont_cls)
+{
+ struct GNUNET_DHT_ControlMessage *msg;
+ struct PendingMessage *pending;
+
+ pending =
+ GNUNET_malloc (sizeof (struct PendingMessage) +
+ sizeof (struct GNUNET_DHT_ControlMessage));
+ msg = (struct GNUNET_DHT_ControlMessage *) &pending[1];
+ pending->msg = &msg->header;
+ msg->header.size = htons (sizeof (struct GNUNET_DHT_ControlMessage));
+ msg->header.type = htons (GNUNET_MESSAGE_TYPE_DHT_CONTROL);
+ msg->command = htons (command);
+ msg->variable = htons (variable);
+ pending->free_on_send = GNUNET_YES;
+ pending->cont = cont;
+ pending->cont_cls = cont_cls;
+ pending->in_pending_queue = GNUNET_YES;
+ GNUNET_CONTAINER_DLL_insert (handle->pending_head, handle->pending_tail,
+ pending);
+ process_pending_messages (handle);
+}
+
+
+/**
+ * Send a message to the DHT telling it to issue a single find
+ * peer request using the peers unique identifier as key. This
+ * is used to fill the routing table, and is normally controlled
+ * by the DHT itself. However, for testing and perhaps more
+ * close control over the DHT, this can be explicitly managed.
+ *
+ * @param handle handle to the DHT service
+ * @param cont continuation to call when done (transmitting request to service)
+ * @param cont_cls closure for cont
+ */
+void
+GNUNET_DHT_find_peers (struct GNUNET_DHT_Handle *handle,
+ GNUNET_SCHEDULER_Task cont, void *cont_cls)
+{
+ send_control_message (handle, GNUNET_MESSAGE_TYPE_DHT_FIND_PEER, 0, cont,
+ cont_cls);
+}
+
+
+
+#if HAVE_MALICIOUS
+
+/**
+ * Send a message to the DHT telling it to start issuing random GET
+ * requests every 'frequency' milliseconds.
+ *
+ * @param handle handle to the DHT service
+ * @param frequency delay between sending malicious messages
+ * @param cont continuation to call when done (transmitting request to service)
+ * @param cont_cls closure for cont
+ */
+void
+GNUNET_DHT_set_malicious_getter (struct GNUNET_DHT_Handle *handle,
+ struct GNUNET_TIME_Relative frequency,
+ GNUNET_SCHEDULER_Task cont, void *cont_cls)
+{
+ if (frequency.rel_value > UINT16_MAX)
+ {
+ GNUNET_break (0);
+ return;
+ }
+ send_control_message (handle, GNUNET_MESSAGE_TYPE_DHT_MALICIOUS_GET,
+ frequency.rel_value, cont, cont_cls);
+}
+
+/**
+ * Send a message to the DHT telling it to start issuing random PUT
+ * requests every 'frequency' milliseconds.
+ *
+ * @param handle handle to the DHT service
+ * @param frequency delay between sending malicious messages
+ * @param cont continuation to call when done (transmitting request to service)
+ * @param cont_cls closure for cont
+ */
+void
+GNUNET_DHT_set_malicious_putter (struct GNUNET_DHT_Handle *handle,
+ struct GNUNET_TIME_Relative frequency,
+ GNUNET_SCHEDULER_Task cont, void *cont_cls)
+{
+ if (frequency.rel_value > UINT16_MAX)
+ {
+ GNUNET_break (0);
+ return;
+ }
+
+ send_control_message (handle, GNUNET_MESSAGE_TYPE_DHT_MALICIOUS_PUT,
+ frequency.rel_value, cont, cont_cls);
+}
+
+
+/**
+ * Send a message to the DHT telling it to start dropping
+ * all requests received.
+ *
+ * @param handle handle to the DHT service
+ * @param cont continuation to call when done (transmitting request to service)
+ * @param cont_cls closure for cont
+ *
+ */
+void
+GNUNET_DHT_set_malicious_dropper (struct GNUNET_DHT_Handle *handle,
+ GNUNET_SCHEDULER_Task cont, void *cont_cls)
+{
+ send_control_message (handle, GNUNET_MESSAGE_TYPE_DHT_MALICIOUS_DROP, 0,
cont,
+ cont_cls);
+}
+
+#endif
+
+/* end of dht_api.c */
Added: gnunet/src/dht/dht_new.h
===================================================================
--- gnunet/src/dht/dht_new.h (rev 0)
+++ gnunet/src/dht/dht_new.h 2011-09-21 05:51:38 UTC (rev 16976)
@@ -0,0 +1,190 @@
+/*
+ This file is part of GNUnet.
+ (C) 2001, 2002, 2003, 2004, 2009, 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
+ 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
+ WITHOUT ANY WARRANTY; without even the implied warranty of
+ MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ General Public License for more details.
+
+ 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.
+*/
+
+/**
+ * @author Christian Grothoff
+ * @author Nathan Evans
+ * @file dht/dht.h
+ */
+
+#ifndef DHT_H_
+#define DHT_H_
+
+
+/**
+ * Message which indicates the DHT should cancel outstanding
+ * requests and discard any state.
+ */
+struct GNUNET_DHT_ClientGetStopMessage
+{
+ /**
+ * Type: GNUNET_MESSAGE_TYPE_DHT_GET_STOP
+ */
+ struct GNUNET_MessageHeader header;
+
+ /**
+ * Always zero.
+ */
+ uint32_t reserved GNUNET_PACKED;
+
+ /**
+ * Unique ID identifying this request
+ */
+ uint64_t unique_id GNUNET_PACKED;
+
+ /**
+ * Key of this request
+ */
+ GNUNET_HashCode key;
+
+};
+
+
+/**
+ * DHT GET message sent from clients to service. Indicates that a GET
+ * request should be issued.
+ */
+struct GNUNET_DHT_ClientGetMessage
+{
+ /**
+ * Type: GNUNET_MESSAGE_TYPE_DHT_CLIENT_GET
+ */
+ struct GNUNET_MessageHeader header;
+
+ /**
+ * Message options, actually an 'enum GNUNET_DHT_RouteOption' value.
+ */
+ uint32_t options GNUNET_PACKED;
+
+ /**
+ * Replication level for this message
+ */
+ uint32_t desired_replication_level GNUNET_PACKED;
+
+ /**
+ * The type for the data for the GET request; actually an 'enum
+ * GNUNET_BLOCK_Type'.
+ */
+ uint32_t type;
+
+ /**
+ * The key to search for
+ */
+ GNUNET_HashCode key;
+
+ /**
+ * Unique ID identifying this request, if 0 then
+ * the client will not expect a response
+ */
+ uint64_t unique_id GNUNET_PACKED;
+
+ /* Possibly followed by xquery, copied to end of this dealy do */
+
+};
+
+
+/**
+ * Reply to a GET send from the service to a client.
+ */
+struct GNUNET_DHT_ClientResultMessage
+{
+ /**
+ * Type: GNUNET_MESSAGE_TYPE_DHT_CLIENT_RESULT
+ */
+ struct GNUNET_MessageHeader header;
+
+ /**
+ * The type for the data.
+ */
+ uint32_t type;
+
+ /**
+ * Number of peers recorded in the outgoing path from source to the
+ * storgage location of this message.
+ */
+ uint32_t put_path_length GNUNET_PACKED;
+
+ /**
+ * The number of peer identities recorded from the storage location
+ * to this peer.
+ */
+ uint32_t get_path_length GNUNET_PACKED;
+
+ /**
+ * Unique ID of the matching GET request.
+ */
+ uint64_t unique_id GNUNET_PACKED;
+
+ /**
+ * When does this entry expire?
+ */
+ struct GNUNET_TIME_AbsoluteNBO expiration;
+
+ /**
+ * The key that was searched for
+ */
+ GNUNET_HashCode key;
+
+ /* put path, get path and actual data are copied to end of this dealy do */
+
+};
+
+
+/**
+ * Message to insert data into the DHT, sent from clients to DHT service.
+ */
+struct GNUNET_DHT_ClientPutMessage
+{
+ /**
+ * Type: GNUNET_MESSAGE_TYPE_DHT_CLIENT_PUT
+ */
+ struct GNUNET_MessageHeader header;
+
+ /**
+ * The type of data to insert.
+ */
+ uint32_t type GNUNET_PACKED;
+
+ /**
+ * Message options, actually an 'enum GNUNET_DHT_RouteOption' value.
+ */
+ uint32_t options GNUNET_PACKED;
+
+ /**
+ * Replication level for this message
+ */
+ uint32_t desired_replication_level GNUNET_PACKED;
+
+ /**
+ * How long should this data persist?
+ */
+ struct GNUNET_TIME_AbsoluteNBO expiration;
+
+ /**
+ * The key to store the value under.
+ */
+ GNUNET_HashCode key;
+
+ /* DATA copied to end of this message */
+
+};
+
+
+#endif
Added: gnunet/src/dht/gnunet-service-dht-new.c
===================================================================
--- gnunet/src/dht/gnunet-service-dht-new.c (rev 0)
+++ gnunet/src/dht/gnunet-service-dht-new.c 2011-09-21 05:51:38 UTC (rev
16976)
@@ -0,0 +1,2991 @@
+/*
+ This file is part of GNUnet.
+ (C) 2009, 2010, 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
+ 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
+ WITHOUT ANY WARRANTY; without even the implied warranty of
+ MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ General Public License for more details.
+
+ 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.
+*/
+
+/**
+ * @file dht/gnunet-service-dht.c
+ * @brief GNUnet DHT service
+ * @author Christian Grothoff
+ * @author Nathan Evans
+ *
+ * TODO:
+ * - use OPTION_MULTIPLE instead of linked list for the forward_list.hashmap
+ * - use different 'struct DHT_MessageContext' for the different types of
+ * messages (currently rather confusing, especially with things like
+ * peer bloom filters occuring when processing replies).
+ */
+
+#include "platform.h"
+#include "gnunet_block_lib.h"
+#include "gnunet_client_lib.h"
+#include "gnunet_getopt_lib.h"
+#include "gnunet_os_lib.h"
+#include "gnunet_protocols.h"
+#include "gnunet_service_lib.h"
+#include "gnunet_nse_service.h"
+#include "gnunet_core_service.h"
+#include "gnunet_signal_lib.h"
+#include "gnunet_util_lib.h"
+#include "gnunet_datacache_lib.h"
+#include "gnunet_transport_service.h"
+#include "gnunet_hello_lib.h"
+#include "gnunet_dht_service.h"
+#include "gnunet_statistics_service.h"
+#include "dht.h"
+#include <fenv.h>
+
+
+/**
+ * Defines whether find peer requests send their HELLO's outgoing,
+ * or expect replies to contain hellos.
+ */
+#define FIND_PEER_WITH_HELLO GNUNET_YES
+
+#define DEFAULT_CORE_QUEUE_SIZE 32
+
+/**
+ * Minimum number of peers we need for "good" routing,
+ * any less than this and we will allow messages to
+ * travel much further through the network!
+ */
+#define MINIMUM_PEER_THRESHOLD 20
+
+/**
+ * Number of requests we track at most (for routing replies).
+ */
+#define DHT_MAX_RECENT (1024 * 16)
+
+/**
+ * How long do we wait at most when queueing messages with core
+ * that we are sending on behalf of other peers.
+ */
+#define DHT_DEFAULT_P2P_TIMEOUT GNUNET_TIME_relative_multiply
(GNUNET_TIME_UNIT_SECONDS, 10)
+
+/**
+ * Default importance for handling messages on behalf of other peers.
+ */
+#define DHT_DEFAULT_P2P_IMPORTANCE 0
+
+/**
+ * How long to keep recent requests around by default.
+ */
+#define DEFAULT_RECENT_REMOVAL GNUNET_TIME_relative_multiply
(GNUNET_TIME_UNIT_SECONDS, 60)
+
+/**
+ * Default time to wait to send find peer messages sent by the dht service.
+ */
+#define DHT_DEFAULT_FIND_PEER_TIMEOUT GNUNET_TIME_relative_multiply
(GNUNET_TIME_UNIT_SECONDS, 30)
+
+/**
+ * Default importance for find peer messages sent by the dht service.
+ */
+#define DHT_DEFAULT_FIND_PEER_IMPORTANCE 8
+
+/**
+ * Default replication parameter for find peer messages sent by the dht
service.
+ */
+#define DHT_DEFAULT_FIND_PEER_REPLICATION 4
+
+/**
+ * How long at least to wait before sending another find peer request.
+ */
+#define DHT_MINIMUM_FIND_PEER_INTERVAL
GNUNET_TIME_relative_multiply(GNUNET_TIME_UNIT_MINUTES, 2)
+
+/**
+ * How long at most to wait before sending another find peer request.
+ */
+#define DHT_MAXIMUM_FIND_PEER_INTERVAL
GNUNET_TIME_relative_multiply(GNUNET_TIME_UNIT_MINUTES, 8)
+
+/**
+ * How often to update our preference levels for peers in our routing tables.
+ */
+#define DHT_DEFAULT_PREFERENCE_INTERVAL
GNUNET_TIME_relative_multiply(GNUNET_TIME_UNIT_MINUTES, 2)
+
+/**
+ * How long at most on average will we allow a reply forward to take
+ * (before we quit sending out new requests)
+ */
+#define MAX_REQUEST_TIME GNUNET_TIME_relative_multiply
(GNUNET_TIME_UNIT_SECONDS, 1)
+
+/**
+ * How many time differences between requesting a core send and
+ * the actual callback to remember.
+ */
+#define MAX_REPLY_TIMES 8
+
+
+
+
+/**
+ * Context containing information about a DHT message received.
+ */
+struct DHT_MessageContext
+{
+ /**
+ * The client this request was received from.
+ * (NULL if received from another peer)
+ */
+ struct ClientList *client;
+
+ /**
+ * The peer this request was received from.
+ */
+ struct GNUNET_PeerIdentity peer;
+
+ /**
+ * Bloomfilter for this routing request.
+ */
+ struct GNUNET_CONTAINER_BloomFilter *bloom;
+
+ /**
+ * extended query (see gnunet_block_lib.h).
+ */
+ const void *xquery;
+
+ /**
+ * Bloomfilter to filter out duplicate replies.
+ */
+ struct GNUNET_CONTAINER_BloomFilter *reply_bf;
+
+ /**
+ * The key this request was about
+ */
+ GNUNET_HashCode key;
+
+ /**
+ * How long should we wait to transmit this request?
+ */
+ struct GNUNET_TIME_Relative timeout;
+
+ /**
+ * The unique identifier of this request
+ */
+ uint64_t unique_id;
+
+ /**
+ * Number of bytes in xquery.
+ */
+ size_t xquery_size;
+
+ /**
+ * Mutator value for the reply_bf, see gnunet_block_lib.h
+ */
+ uint32_t reply_bf_mutator;
+
+ /**
+ * Desired replication level
+ */
+ uint32_t replication;
+
+ /**
+ * Network size estimate, either ours or the sum of
+ * those routed to thus far. =~ Log of number of peers
+ * chosen from for this request.
+ */
+ uint32_t network_size;
+
+ /**
+ * Any message options for this request
+ */
+ uint32_t msg_options;
+
+ /**
+ * How many hops has the message already traversed?
+ */
+ uint32_t hop_count;
+
+ /**
+ * How many peer identities are present in the path history?
+ */
+ uint32_t path_history_len;
+
+ /**
+ * Path history.
+ */
+ char *path_history;
+
+ /**
+ * How important is this message?
+ */
+ unsigned int importance;
+
+ /**
+ * Should we (still) forward the request on to other peers?
+ */
+ int do_forward;
+
+ /**
+ * Did we forward this message? (may need to remember it!)
+ */
+ int forwarded;
+
+ /**
+ * Are we the closest known peer to this key (out of our neighbors?)
+ */
+ int closest;
+};
+
+
+/**
+ * Record used for remembering what peers are waiting for what
+ * responses (based on search key).
+ */
+struct DHTRouteSource
+{
+ /**
+ * This is a DLL.
+ */
+ struct DHTRouteSource *next;
+
+ /**
+ * This is a DLL.
+ */
+ struct DHTRouteSource *prev;
+
+ /**
+ * UID of the request, 0 if from another peer.
+ */
+ uint64_t uid;
+
+ /**
+ * Source of the request. Replies should be forwarded to
+ * this peer.
+ */
+ struct GNUNET_PeerIdentity source;
+
+ /**
+ * If this was a local request, remember the client; otherwise NULL.
+ */
+ struct ClientList *client;
+
+ /**
+ * Pointer to this nodes heap location (for removal)
+ */
+ struct GNUNET_CONTAINER_HeapNode *hnode;
+
+ /**
+ * Back pointer to the record storing this information.
+ */
+ struct DHTQueryRecord *record;
+
+ /**
+ * Task to remove this entry on timeout.
+ */
+ GNUNET_SCHEDULER_TaskIdentifier delete_task;
+
+ /**
+ * Bloomfilter of peers we have already sent back as
+ * replies to the initial request. Allows us to not
+ * forward the same peer multiple times for a find peer
+ * request.
+ */
+ struct GNUNET_CONTAINER_BloomFilter *find_peers_responded;
+
+};
+
+
+/**
+ * Entry in the DHT routing table.
+ */
+struct DHTQueryRecord
+{
+ /**
+ * Head of DLL for result forwarding.
+ */
+ struct DHTRouteSource *head;
+
+ /**
+ * Tail of DLL for result forwarding.
+ */
+ struct DHTRouteSource *tail;
+
+ /**
+ * Key that the record concerns.
+ */
+ GNUNET_HashCode key;
+
+};
+
+
+/**
+ * Context used to calculate the number of find peer messages
+ * per X time units since our last scheduled find peer message
+ * was sent. If we have seen too many messages, delay or don't
+ * send our own out.
+ */
+struct FindPeerMessageContext
+{
+ unsigned int count;
+
+ struct GNUNET_TIME_Absolute start;
+
+};
+
+
+struct RecentRequest
+{
+ /**
+ * Position of this node in the min heap.
+ */
+ struct GNUNET_CONTAINER_HeapNode *heap_node;
+
+ /**
+ * Bloomfilter containing entries for peers
+ * we forwarded this request to.
+ */
+ struct GNUNET_CONTAINER_BloomFilter *bloom;
+
+ /**
+ * Timestamp of this request, for ordering
+ * the min heap.
+ */
+ struct GNUNET_TIME_Absolute timestamp;
+
+ /**
+ * Key of this request.
+ */
+ GNUNET_HashCode key;
+
+ /**
+ * Unique identifier for this request, 0 if from another peer.
+ */
+ uint64_t uid;
+
+ /**
+ * Task to remove this entry on timeout.
+ */
+ GNUNET_SCHEDULER_TaskIdentifier remove_task;
+};
+
+
+/**
+ * Recent requests by time inserted.
+ */
+static struct GNUNET_CONTAINER_Heap *recent_heap;
+
+/**
+ * Context to use to calculate find peer rates.
+ */
+static struct FindPeerMessageContext find_peer_context;
+
+/**
+ * How many peers have we added since we sent out our last
+ * find peer request?
+ */
+static unsigned int newly_found_peers;
+
+/**
+ * Handle to the datacache service (for inserting/retrieving data)
+ */
+static struct GNUNET_DATACACHE_Handle *datacache;
+
+/**
+ * Handle for the statistics service.
+ */
+struct GNUNET_STATISTICS_Handle *stats;
+
+/**
+ * Handle to get our current HELLO.
+ */
+static struct GNUNET_TRANSPORT_GetHelloHandle *ghh;
+
+/**
+ * The configuration the DHT service is running with
+ */
+static const struct GNUNET_CONFIGURATION_Handle *cfg;
+
+/**
+ * Handle to the core service
+ */
+static struct GNUNET_CORE_Handle *coreAPI;
+
+/**
+ * Handle to the transport service, for getting our hello
+ */
+static struct GNUNET_TRANSPORT_Handle *transport_handle;
+
+/**
+ * The identity of our peer.
+ */
+static struct GNUNET_PeerIdentity my_identity;
+
+/**
+ * Short id of the peer, for printing
+ */
+static char *my_short_id;
+
+/**
+ * Our HELLO
+ */
+static struct GNUNET_MessageHeader *my_hello;
+
+/**
+ * Task to run when we shut down, cleaning up all our trash
+ */
+static GNUNET_SCHEDULER_TaskIdentifier cleanup_task;
+
+/**
+ * Recently seen find peer requests.
+ */
+static struct GNUNET_CONTAINER_MultiHashMap *recent_find_peer_requests;
+
+/**
+ * Reply times for requests, if we are busy, don't send any
+ * more requests!
+ */
+static struct GNUNET_TIME_Relative reply_times[MAX_REPLY_TIMES];
+
+/**
+ * Current counter for replies.
+ */
+static unsigned int reply_counter;
+
+/**
+ * Our handle to the BLOCK library.
+ */
+static struct GNUNET_BLOCK_Context *block_context;
+
+
+
+/** Declare here so retry_core_send is aware of it */
+static size_t
+core_transmit_notify (void *cls, size_t size, void *buf);
+
+
+
+/**
+ * Given the largest send delay, artificially decrease it
+ * so the next time around we may have a chance at sending
+ * again.
+ */
+static void
+decrease_max_send_delay (struct GNUNET_TIME_Relative max_time)
+{
+ unsigned int i;
+
+ for (i = 0; i < MAX_REPLY_TIMES; i++)
+ {
+ if (reply_times[i].rel_value == max_time.rel_value)
+ {
+ reply_times[i].rel_value = reply_times[i].rel_value / 2;
+ return;
+ }
+ }
+}
+
+
+/**
+ * Find the maximum send time of the recently sent values.
+ *
+ * @return the average time between asking core to send a message
+ * and when the buffer for copying it is passed
+ */
+static struct GNUNET_TIME_Relative
+get_max_send_delay ()
+{
+ unsigned int i;
+ struct GNUNET_TIME_Relative max_time;
+
+ max_time = GNUNET_TIME_relative_get_zero ();
+
+ for (i = 0; i < MAX_REPLY_TIMES; i++)
+ {
+ if (reply_times[i].rel_value > max_time.rel_value)
+ max_time.rel_value = reply_times[i].rel_value;
+ }
+#if DEBUG_DHT
+ if (max_time.rel_value > MAX_REQUEST_TIME.rel_value)
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Max send delay was %llu\n",
+ (unsigned long long) max_time.rel_value);
+#endif
+ return max_time;
+}
+
+
+static void
+increment_stats (const char *value)
+{
+ if (stats == NULL)
+ return;
+ GNUNET_STATISTICS_update (stats, value, 1, GNUNET_NO);
+}
+
+
+static void
+decrement_stats (const char *value)
+{
+ if (stats == NULL)
+ return;
+ GNUNET_STATISTICS_update (stats, value, -1, GNUNET_NO);
+}
+
+
+/**
+ * Try to send another message from our core send list
+ */
+static void
+try_core_send (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
+{
+ struct PeerInfo *peer = cls;
+ struct P2PPendingMessage *pending;
+ size_t ssize;
+
+ peer->send_task = GNUNET_SCHEDULER_NO_TASK;
+
+ if ((tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN) != 0)
+ return;
+
+ if (peer->th != NULL)
+ return; /* Message send already in progress */
+
+ pending = peer->head;
+ if (pending != NULL)
+ {
+ ssize = ntohs (pending->msg->size);
+#if DEBUG_DHT > 1
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "`%s:%s': Calling notify_transmit_ready with size %d for peer
%s\n",
+ my_short_id, "DHT", ssize, GNUNET_i2s (&peer->id));
+#endif
+ pending->scheduled = GNUNET_TIME_absolute_get ();
+ reply_counter++;
+ if (reply_counter >= MAX_REPLY_TIMES)
+ reply_counter = 0;
+ peer->th =
+ GNUNET_CORE_notify_transmit_ready (coreAPI, GNUNET_YES,
+ pending->importance,
+ pending->timeout, &peer->id, ssize,
+ &core_transmit_notify, peer);
+ if (peer->th == NULL)
+ increment_stats ("# notify transmit ready failed");
+ }
+}
+
+
+/**
+ * Function called to send a request out to another peer.
+ * Called both for locally initiated requests and those
+ * received from other peers.
+ *
+ * @param msg the encapsulated message
+ * @param peer the peer to forward the message to
+ * @param msg_ctx the context of the message (hop count, bloom, etc.)
+ */
+static void
+forward_result_message (const struct GNUNET_MessageHeader *msg,
+ struct PeerInfo *peer,
+ struct DHT_MessageContext *msg_ctx)
+{
+ struct GNUNET_DHT_P2PRouteResultMessage *result_message;
+ struct P2PPendingMessage *pending;
+ size_t msize;
+ size_t psize;
+ char *path_start;
+ char *path_offset;
+
+ increment_stats (STAT_RESULT_FORWARDS);
+ msize =
+ sizeof (struct GNUNET_DHT_P2PRouteResultMessage) + ntohs (msg->size) +
+ (sizeof (struct GNUNET_PeerIdentity) * msg_ctx->path_history_len);
+ GNUNET_assert (msize <= GNUNET_SERVER_MAX_MESSAGE_SIZE);
+ psize = sizeof (struct P2PPendingMessage) + msize;
+ pending = GNUNET_malloc (psize);
+ pending->msg = (struct GNUNET_MessageHeader *) &pending[1];
+ pending->importance = DHT_SEND_PRIORITY;
+ pending->timeout = GNUNET_TIME_relative_get_forever ();
+ result_message = (struct GNUNET_DHT_P2PRouteResultMessage *) pending->msg;
+ result_message->header.size = htons (msize);
+ result_message->header.type =
+ htons (GNUNET_MESSAGE_TYPE_DHT_P2P_ROUTE_RESULT);
+ result_message->outgoing_path_length = htonl (msg_ctx->path_history_len);
+ if (msg_ctx->path_history_len > 0)
+ {
+ /* End of pending is where enc_msg starts */
+ path_start = (char *) &pending[1];
+ /* Offset by the size of the enc_msg */
+ path_start += ntohs (msg->size);
+ memcpy (path_start, msg_ctx->path_history,
+ msg_ctx->path_history_len * (sizeof (struct GNUNET_PeerIdentity)));
+ }
+ result_message->options = htonl (msg_ctx->msg_options);
+ result_message->hop_count = htonl (msg_ctx->hop_count + 1);
+ memcpy (&result_message->key, &msg_ctx->key, sizeof (GNUNET_HashCode));
+ /* Copy the enc_msg, then the path history as well! */
+ memcpy (&result_message[1], msg, ntohs (msg->size));
+ path_offset = (char *) &result_message[1];
+ path_offset += ntohs (msg->size);
+ /* If we have path history, copy it to the end of the whole thing */
+ if (msg_ctx->path_history_len > 0)
+ memcpy (path_offset, msg_ctx->path_history,
+ msg_ctx->path_history_len * (sizeof (struct GNUNET_PeerIdentity)));
+#if DEBUG_DHT > 1
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "%s:%s Adding pending message size %d for peer %s\n",
my_short_id,
+ "DHT", msize, GNUNET_i2s (&peer->id));
+#endif
+ peer->pending_count++;
+ increment_stats ("# pending messages scheduled");
+ GNUNET_CONTAINER_DLL_insert_after (peer->head, peer->tail, peer->tail,
+ pending);
+ if (peer->send_task == GNUNET_SCHEDULER_NO_TASK)
+ peer->send_task = GNUNET_SCHEDULER_add_now (&try_core_send, peer);
+}
+
+
+/**
+ * Called when core is ready to send a message we asked for
+ * out to the destination.
+ *
+ * @param cls closure (NULL)
+ * @param size number of bytes available in buf
+ * @param buf where the callee should write the message
+ * @return number of bytes written to buf
+ */
+static size_t
+core_transmit_notify (void *cls, size_t size, void *buf)
+{
+ struct PeerInfo *peer = cls;
+ char *cbuf = buf;
+ struct P2PPendingMessage *pending;
+
+ size_t off;
+ size_t msize;
+
+ peer->th = NULL;
+ if (buf == NULL)
+ {
+ /* client disconnected */
+#if DEBUG_DHT
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "`%s:%s': buffer was NULL\n",
+ my_short_id, "DHT");
+#endif
+ return 0;
+ }
+
+ if (peer->head == NULL)
+ return 0;
+
+ off = 0;
+ pending = peer->head;
+ while (NULL != pending &&
+ (size - off >= (msize = ntohs (pending->msg->size))))
+ {
+ memcpy (&cbuf[off], pending->msg, msize);
+ off += msize;
+ peer->pending_count--;
+ increment_stats ("# pending messages sent");
+ GNUNET_CONTAINER_DLL_remove (peer->head, peer->tail, pending);
+ GNUNET_free (pending);
+ pending = peer->head;
+ }
+ if ((peer->head != NULL) && (peer->send_task == GNUNET_SCHEDULER_NO_TASK))
+ peer->send_task = GNUNET_SCHEDULER_add_now (&try_core_send, peer);
+
+ return off;
+}
+
+
+/**
+ * Compute the distance between have and target as a 32-bit value.
+ * Differences in the lower bits must count stronger than differences
+ * in the higher bits.
+ *
+ * @return 0 if have==target, otherwise a number
+ * that is larger as the distance between
+ * the two hash codes increases
+ */
+static unsigned int
+distance (const GNUNET_HashCode * target, const GNUNET_HashCode * have)
+{
+ unsigned int bucket;
+ unsigned int msb;
+ unsigned int lsb;
+ unsigned int i;
+
+ /* We have to represent the distance between two 2^9 (=512)-bit
+ * numbers as a 2^5 (=32)-bit number with "0" being used for the
+ * two numbers being identical; furthermore, we need to
+ * guarantee that a difference in the number of matching
+ * bits is always represented in the result.
+ *
+ * We use 2^32/2^9 numerical values to distinguish between
+ * hash codes that have the same LSB bit distance and
+ * use the highest 2^9 bits of the result to signify the
+ * number of (mis)matching LSB bits; if we have 0 matching
+ * and hence 512 mismatching LSB bits we return -1 (since
+ * 512 itself cannot be represented with 9 bits) */
+
+ /* first, calculate the most significant 9 bits of our
+ * result, aka the number of LSBs */
+ bucket = GNUNET_CRYPTO_hash_matching_bits (target, have);
+ /* bucket is now a value between 0 and 512 */
+ if (bucket == 512)
+ return 0; /* perfect match */
+ if (bucket == 0)
+ return (unsigned int) -1; /* LSB differs; use max (if we did the
bit-shifting
+ * below, we'd end up with max+1 (overflow)) */
+
+ /* calculate the most significant bits of the final result */
+ msb = (512 - bucket) << (32 - 9);
+ /* calculate the 32-9 least significant bits of the final result by
+ * looking at the differences in the 32-9 bits following the
+ * mismatching bit at 'bucket' */
+ lsb = 0;
+ for (i = bucket + 1;
+ (i < sizeof (GNUNET_HashCode) * 8) && (i < bucket + 1 + 32 - 9); i++)
+ {
+ if (GNUNET_CRYPTO_hash_get_bit (target, i) !=
+ GNUNET_CRYPTO_hash_get_bit (have, i))
+ lsb |= (1 << (bucket + 32 - 9 - i)); /* first bit set will be 10,
+ * last bit set will be 31 --
if
+ * i does not reach 512
first... */
+ }
+ return msb | lsb;
+}
+
+
+/**
+ * Return a number that is larger the closer the
+ * "have" GNUNET_hash code is to the "target".
+ *
+ * @return inverse distance metric, non-zero.
+ * Must fudge the value if NO bits match.
+ */
+static unsigned int
+inverse_distance (const GNUNET_HashCode * target, const GNUNET_HashCode * have)
+{
+ if (GNUNET_CRYPTO_hash_matching_bits (target, have) == 0)
+ return 1; /* Never return 0! */
+ return ((unsigned int) -1) - distance (target, have);
+}
+
+
+/**
+ * Find the optimal bucket for this key, regardless
+ * of the current number of buckets in use.
+ *
+ * @param hc the hashcode to compare our identity to
+ *
+ * @return the proper bucket index, or GNUNET_SYSERR
+ * on error (same hashcode)
+ */
+static int
+find_bucket (const GNUNET_HashCode * hc)
+{
+ unsigned int bits;
+
+ bits = GNUNET_CRYPTO_hash_matching_bits (&my_identity.hashPubKey, hc);
+ if (bits == MAX_BUCKETS)
+ return GNUNET_SYSERR;
+ return MAX_BUCKETS - bits - 1;
+}
+
+
+/**
+ * Find which k-bucket this peer should go into,
+ * taking into account the size of the k-bucket
+ * array. This means that if more bits match than
+ * there are currently buckets, lowest_bucket will
+ * be returned.
+ *
+ * @param hc GNUNET_HashCode we are finding the bucket for.
+ *
+ * @return the proper bucket index for this key,
+ * or GNUNET_SYSERR on error (same hashcode)
+ */
+static int
+find_current_bucket (const GNUNET_HashCode * hc)
+{
+ int actual_bucket;
+
+ actual_bucket = find_bucket (hc);
+ if (actual_bucket == GNUNET_SYSERR) /* hc and our peer identity match! */
+ return lowest_bucket;
+ if (actual_bucket < lowest_bucket) /* actual_bucket not yet used */
+ return lowest_bucket;
+ return actual_bucket;
+}
+
+
+/**
+ * Find a routing table entry from a peer identity
+ *
+ * @param peer the peer identity to look up
+ *
+ * @return the routing table entry, or NULL if not found
+ */
+static struct PeerInfo *
+find_peer_by_id (const struct GNUNET_PeerIdentity *peer)
+{
+ int bucket;
+ struct PeerInfo *pos;
+
+ bucket = find_current_bucket (&peer->hashPubKey);
+
+ if (0 == memcmp (&my_identity, peer, sizeof (struct GNUNET_PeerIdentity)))
+ return NULL;
+
+ pos = k_buckets[bucket].head;
+ while (pos != NULL)
+ {
+ if (0 == memcmp (&pos->id, peer, sizeof (struct GNUNET_PeerIdentity)))
+ return pos;
+ pos = pos->next;
+ }
+ return NULL; /* No such peer. */
+}
+
+/* Forward declaration */
+static void
+update_core_preference (void *cls,
+ const struct GNUNET_SCHEDULER_TaskContext *tc);
+
+
+/**
+ * Function called with statistics about the given peer.
+ *
+ * @param cls closure
+ * @param peer identifies the peer
+ * @param bpm_out set to the current bandwidth limit (sending) for this peer
+ * @param amount set to the amount that was actually reserved or unreserved;
+ * either the full requested amount or zero (no partial
reservations)
+ * @param res_delay if the reservation could not be satisfied (amount was 0),
how
+ * long should the client wait until re-trying?
+ * @param preference current traffic preference for the given peer
+ */
+static void
+update_core_preference_finish (void *cls,
+ const struct GNUNET_PeerIdentity *peer,
+ struct GNUNET_BANDWIDTH_Value32NBO bpm_out,
+ int32_t amount,
+ struct GNUNET_TIME_Relative res_delay,
+ uint64_t preference)
+{
+ struct PeerInfo *peer_info = cls;
+
+ peer_info->info_ctx = NULL;
+ GNUNET_SCHEDULER_add_delayed (DHT_DEFAULT_PREFERENCE_INTERVAL,
+ &update_core_preference, peer_info);
+}
+
+static void
+update_core_preference (void *cls,
+ const struct GNUNET_SCHEDULER_TaskContext *tc)
+{
+ struct PeerInfo *peer = cls;
+ uint64_t preference;
+ unsigned int matching;
+
+ if ((tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN) != 0)
+ {
+ return;
+ }
+ matching =
+ GNUNET_CRYPTO_hash_matching_bits (&my_identity.hashPubKey,
+ &peer->id.hashPubKey);
+ if (matching >= 64)
+ {
+#if DEBUG_DHT
+ GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+ "Peer identifier matches by %u bits, only shifting as much as
we can!\n",
+ matching);
+#endif
+ matching = 63;
+ }
+ preference = 1LL << matching;
+ peer->info_ctx =
+ GNUNET_CORE_peer_change_preference (coreAPI, &peer->id,
+ GNUNET_TIME_UNIT_FOREVER_REL,
+ GNUNET_BANDWIDTH_VALUE_MAX, 0,
+ preference,
+ &update_core_preference_finish,
peer);
+}
+
+
+/**
+ * Given a peer and its corresponding bucket,
+ * remove it from that bucket. Does not free
+ * the PeerInfo struct, nor cancel messages
+ * or free messages waiting to be sent to this
+ * peer!
+ *
+ * @param peer the peer to remove
+ * @param bucket the bucket the peer belongs to
+ */
+static void
+remove_peer (struct PeerInfo *peer, unsigned int bucket)
+{
+ GNUNET_assert (k_buckets[bucket].peers_size > 0);
+ GNUNET_CONTAINER_DLL_remove (k_buckets[bucket].head, k_buckets[bucket].tail,
+ peer);
+ k_buckets[bucket].peers_size--;
+ if ((bucket == lowest_bucket) && (k_buckets[lowest_bucket].peers_size == 0)
&&
+ (lowest_bucket < MAX_BUCKETS - 1))
+ lowest_bucket++;
+}
+
+/**
+ * Removes peer from a bucket, then frees associated
+ * resources and frees peer.
+ *
+ * @param peer peer to be removed and freed
+ * @param bucket which bucket this peer belongs to
+ */
+static void
+delete_peer (struct PeerInfo *peer, unsigned int bucket)
+{
+ struct P2PPendingMessage *pos;
+ struct P2PPendingMessage *next;
+
+ remove_peer (peer, bucket); /* First remove the peer from its bucket */
+ if (peer->send_task != GNUNET_SCHEDULER_NO_TASK)
+ GNUNET_SCHEDULER_cancel (peer->send_task);
+ if ((peer->th != NULL) && (coreAPI != NULL))
+ GNUNET_CORE_notify_transmit_ready_cancel (peer->th);
+
+ pos = peer->head;
+ while (pos != NULL) /* Remove any pending messages for this peer */
+ {
+ increment_stats
+ ("# dht pending messages discarded (due to disconnect/shutdown)");
+ next = pos->next;
+ GNUNET_free (pos);
+ pos = next;
+ }
+
+ GNUNET_assert (GNUNET_CONTAINER_multihashmap_contains
+ (all_known_peers, &peer->id.hashPubKey));
+ GNUNET_assert (GNUNET_YES ==
+ GNUNET_CONTAINER_multihashmap_remove (all_known_peers,
+ &peer->id.hashPubKey,
+ peer));
+ GNUNET_free (peer);
+ decrement_stats (STAT_PEERS_KNOWN);
+}
+
+
+/**
+ * Iterator over hash map entries.
+ *
+ * @param cls closure
+ * @param key current key code
+ * @param value PeerInfo of the peer to move to new lowest bucket
+ * @return GNUNET_YES if we should continue to
+ * iterate,
+ * GNUNET_NO if not.
+ */
+static int
+move_lowest_bucket (void *cls, const GNUNET_HashCode * key, void *value)
+{
+ struct PeerInfo *peer = value;
+ int new_bucket;
+
+ GNUNET_assert (lowest_bucket > 0);
+ new_bucket = lowest_bucket - 1;
+ remove_peer (peer, lowest_bucket);
+ GNUNET_CONTAINER_DLL_insert_after (k_buckets[new_bucket].head,
+ k_buckets[new_bucket].tail,
+ k_buckets[new_bucket].tail, peer);
+ k_buckets[new_bucket].peers_size++;
+ return GNUNET_YES;
+}
+
+
+/**
+ * The current lowest bucket is full, so change the lowest
+ * bucket to the next lower down, and move any appropriate
+ * entries in the current lowest bucket to the new bucket.
+ */
+static void
+enable_next_bucket ()
+{
+ struct GNUNET_CONTAINER_MultiHashMap *to_remove;
+ struct PeerInfo *pos;
+
+ GNUNET_assert (lowest_bucket > 0);
+ to_remove = GNUNET_CONTAINER_multihashmap_create (bucket_size);
+ pos = k_buckets[lowest_bucket].head;
+
+ /* Populate the array of peers which should be in the next lowest bucket */
+ while (pos != NULL)
+ {
+ if (find_bucket (&pos->id.hashPubKey) < lowest_bucket)
+ GNUNET_CONTAINER_multihashmap_put (to_remove, &pos->id.hashPubKey, pos,
+
GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY);
+ pos = pos->next;
+ }
+
+ /* Remove peers from lowest bucket, insert into next lowest bucket */
+ GNUNET_CONTAINER_multihashmap_iterate (to_remove, &move_lowest_bucket, NULL);
+ GNUNET_CONTAINER_multihashmap_destroy (to_remove);
+ lowest_bucket = lowest_bucket - 1;
+}
+
+
+/**
+ * Find the closest peer in our routing table to the
+ * given hashcode.
+ *
+ * @return The closest peer in our routing table to the
+ * key, or NULL on error.
+ */
+static struct PeerInfo *
+find_closest_peer (const GNUNET_HashCode * hc)
+{
+ struct PeerInfo *pos;
+ struct PeerInfo *current_closest;
+ unsigned int lowest_distance;
+ unsigned int temp_distance;
+ int bucket;
+ int count;
+
+ lowest_distance = -1;
+
+ if (k_buckets[lowest_bucket].peers_size == 0)
+ return NULL;
+
+ current_closest = NULL;
+ for (bucket = lowest_bucket; bucket < MAX_BUCKETS; bucket++)
+ {
+ pos = k_buckets[bucket].head;
+ count = 0;
+ while ((pos != NULL) && (count < bucket_size))
+ {
+ temp_distance = distance (&pos->id.hashPubKey, hc);
+ if (temp_distance <= lowest_distance)
+ {
+ lowest_distance = temp_distance;
+ current_closest = pos;
+ }
+ pos = pos->next;
+ count++;
+ }
+ }
+ GNUNET_assert (current_closest != NULL);
+ return current_closest;
+}
+
+
+/**
+ * Function called to send a request out to another peer.
+ * Called both for locally initiated requests and those
+ * received from other peers.
+ *
+ * @param msg the encapsulated message
+ * @param peer the peer to forward the message to
+ * @param msg_ctx the context of the message (hop count, bloom, etc.)
+ */
+static void
+forward_message (const struct GNUNET_MessageHeader *msg, struct PeerInfo *peer,
+ struct DHT_MessageContext *msg_ctx)
+{
+ struct GNUNET_DHT_P2PRouteMessage *route_message;
+ struct P2PPendingMessage *pending;
+ size_t msize;
+ size_t psize;
+ char *route_path;
+
+ increment_stats (STAT_ROUTE_FORWARDS);
+ GNUNET_assert (peer != NULL);
+ if ((msg_ctx->closest != GNUNET_YES) &&
+ (peer == find_closest_peer (&msg_ctx->key)))
+ increment_stats (STAT_ROUTE_FORWARDS_CLOSEST);
+
+ msize =
+ sizeof (struct GNUNET_DHT_P2PRouteMessage) + ntohs (msg->size) +
+ (msg_ctx->path_history_len * sizeof (struct GNUNET_PeerIdentity));
+ GNUNET_assert (msize <= GNUNET_SERVER_MAX_MESSAGE_SIZE);
+ psize = sizeof (struct P2PPendingMessage) + msize;
+ pending = GNUNET_malloc (psize);
+ pending->msg = (struct GNUNET_MessageHeader *) &pending[1];
+ pending->importance = msg_ctx->importance;
+ pending->timeout = msg_ctx->timeout;
+ route_message = (struct GNUNET_DHT_P2PRouteMessage *) pending->msg;
+ route_message->header.size = htons (msize);
+ route_message->header.type = htons (GNUNET_MESSAGE_TYPE_DHT_P2P_ROUTE);
+ route_message->options = htonl (msg_ctx->msg_options);
+ route_message->hop_count = htonl (msg_ctx->hop_count + 1);
+ route_message->network_size = htonl (msg_ctx->network_size);
+ route_message->desired_replication_level = htonl (msg_ctx->replication);
+ if (msg_ctx->bloom != NULL)
+ GNUNET_assert (GNUNET_OK ==
+ GNUNET_CONTAINER_bloomfilter_get_raw_data (msg_ctx->bloom,
+ route_message->
+ bloomfilter,
+ DHT_BLOOM_SIZE));
+ memcpy (&route_message->key, &msg_ctx->key, sizeof (GNUNET_HashCode));
+ memcpy (&route_message[1], msg, ntohs (msg->size));
+ if (GNUNET_DHT_RO_RECORD_ROUTE ==
+ (msg_ctx->msg_options & GNUNET_DHT_RO_RECORD_ROUTE))
+ {
+ route_message->outgoing_path_length = htonl (msg_ctx->path_history_len);
+ /* Set pointer to start of enc_msg */
+ route_path = (char *) &route_message[1];
+ /* Offset to the end of the enc_msg */
+ route_path += ntohs (msg->size);
+ /* Copy the route_path after enc_msg */
+ memcpy (route_path, msg_ctx->path_history,
+ msg_ctx->path_history_len * sizeof (struct GNUNET_PeerIdentity));
+ }
+#if DEBUG_DHT > 1
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "%s:%s Adding pending message size %d for peer %s\n",
my_short_id,
+ "DHT", msize, GNUNET_i2s (&peer->id));
+#endif
+ peer->pending_count++;
+ increment_stats ("# pending messages scheduled");
+ GNUNET_CONTAINER_DLL_insert_after (peer->head, peer->tail, peer->tail,
+ pending);
+ if (peer->send_task == GNUNET_SCHEDULER_NO_TASK)
+ peer->send_task = GNUNET_SCHEDULER_add_now (&try_core_send, peer);
+}
+
+
+
+
+/**
+ * Called when a reply needs to be sent to a client, as
+ * a result it found to a GET or FIND PEER request.
+ *
+ * @param client the client to send the reply to
+ * @param message the encapsulated message to send
+ * @param msg_ctx the context of the received message
+ */
+static void
+send_reply_to_client (struct ClientList *client,
+ const struct GNUNET_MessageHeader *message,
+ struct DHT_MessageContext *msg_ctx)
+{
+ struct GNUNET_DHT_RouteResultMessage *reply;
+ struct PendingMessage *pending_message;
+ uint16_t msize;
+ size_t tsize;
+ char *reply_offset;
+
+#if DEBUG_DHT
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "`%s:%s': Sending reply to client.\n",
+ my_short_id, "DHT");
+#endif
+ msize = ntohs (message->size);
+ tsize =
+ sizeof (struct GNUNET_DHT_RouteResultMessage) + msize +
+ (msg_ctx->path_history_len * sizeof (struct GNUNET_PeerIdentity));
+ if (tsize >= GNUNET_SERVER_MAX_MESSAGE_SIZE)
+ {
+ GNUNET_break_op (0);
+ return;
+ }
+ pending_message = GNUNET_malloc (sizeof (struct PendingMessage) + tsize);
+ pending_message->msg = (struct GNUNET_MessageHeader *) &pending_message[1];
+ reply = (struct GNUNET_DHT_RouteResultMessage *) &pending_message[1];
+ reply->header.type = htons (GNUNET_MESSAGE_TYPE_DHT_LOCAL_ROUTE_RESULT);
+ reply->header.size = htons (tsize);
+ reply->outgoing_path_length = htonl (msg_ctx->path_history_len);
+ reply->unique_id = GNUNET_htonll (msg_ctx->unique_id);
+ memcpy (&reply->key, &msg_ctx->key, sizeof (GNUNET_HashCode));
+ reply_offset = (char *) &reply[1];
+ memcpy (&reply[1], message, msize);
+ if (msg_ctx->path_history_len > 0)
+ {
+ reply_offset += msize;
+ memcpy (reply_offset, msg_ctx->path_history,
+ msg_ctx->path_history_len * sizeof (struct GNUNET_PeerIdentity));
+ }
+ add_pending_message (client, pending_message);
+}
+
+/**
+ * Consider whether or not we would like to have this peer added to
+ * our routing table. Check whether bucket for this peer is full,
+ * if so return negative; if not return positive. Since peers are
+ * only added on CORE level connect, this doesn't actually add the
+ * peer to the routing table.
+ *
+ * @param peer the peer we are considering adding
+ *
+ * @return GNUNET_YES if we want this peer, GNUNET_NO if not (bucket
+ * already full)
+ */
+static int
+consider_peer (struct GNUNET_PeerIdentity *peer)
+{
+ int bucket;
+
+ if ((GNUNET_YES ==
+ GNUNET_CONTAINER_multihashmap_contains (all_known_peers,
+ &peer->hashPubKey)) ||
+ (0 == memcmp (&my_identity, peer, sizeof (struct GNUNET_PeerIdentity))))
+ return GNUNET_NO; /* We already know this peer (are connected
even!) */
+ bucket = find_current_bucket (&peer->hashPubKey);
+
+ if ((k_buckets[bucket].peers_size < bucket_size) ||
+ ((bucket == lowest_bucket) && (lowest_bucket > 0)))
+ return GNUNET_YES;
+
+ return GNUNET_NO;
+}
+
+
+/**
+ * Task used to remove forwarding entries, either
+ * after timeout, when full, or on shutdown.
+ *
+ * @param cls the entry to remove
+ * @param tc context, reason, etc.
+ */
+static void
+remove_forward_entry (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
+{
+ struct DHTRouteSource *source_info = cls;
+ struct DHTQueryRecord *record;
+
+ source_info = GNUNET_CONTAINER_heap_remove_node (source_info->hnode);
+ record = source_info->record;
+ GNUNET_CONTAINER_DLL_remove (record->head, record->tail, source_info);
+
+ if (record->head == NULL) /* No more entries in DLL */
+ {
+ GNUNET_assert (GNUNET_YES ==
+ GNUNET_CONTAINER_multihashmap_remove (forward_list.hashmap,
+ &record->key,
record));
+ GNUNET_free (record);
+ }
+ if (source_info->find_peers_responded != NULL)
+ GNUNET_CONTAINER_bloomfilter_free (source_info->find_peers_responded);
+ GNUNET_free (source_info);
+}
+
+/**
+ * Main function that handles whether or not to route a result
+ * message to other peers, or to send to our local client.
+ *
+ * @param msg the result message to be routed
+ * @param msg_ctx context of the message we are routing
+ *
+ * @return the number of peers the message was routed to,
+ * GNUNET_SYSERR on failure
+ */
+static int
+route_result_message (struct GNUNET_MessageHeader *msg,
+ struct DHT_MessageContext *msg_ctx)
+{
+ struct GNUNET_PeerIdentity new_peer;
+ struct DHTQueryRecord *record;
+ struct DHTRouteSource *pos;
+ struct PeerInfo *peer_info;
+ const struct GNUNET_MessageHeader *hello_msg;
+
+#if DEBUG_DHT > 1
+ unsigned int i;
+#endif
+
+ increment_stats (STAT_RESULTS);
+ /**
+ * If a find peer result message is received and contains a valid
+ * HELLO for another peer, offer it to the transport service.
+ */
+ if (ntohs (msg->type) == GNUNET_MESSAGE_TYPE_DHT_FIND_PEER_RESULT)
+ {
+ if (ntohs (msg->size) <= sizeof (struct GNUNET_MessageHeader))
+ GNUNET_break_op (0);
+
+ hello_msg = &msg[1];
+ if ((ntohs (hello_msg->type) != GNUNET_MESSAGE_TYPE_HELLO) ||
+ (GNUNET_SYSERR ==
+ GNUNET_HELLO_get_id ((const struct GNUNET_HELLO_Message *) hello_msg,
+ &new_peer)))
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+ "%s:%s Received non-HELLO message type in find peer result
message!\n",
+ my_short_id, "DHT");
+ GNUNET_break_op (0);
+ return GNUNET_NO;
+ }
+ else /* We have a valid hello, and peer id stored
in new_peer */
+ {
+ find_peer_context.count++;
+ increment_stats (STAT_FIND_PEER_REPLY);
+ if (GNUNET_YES == consider_peer (&new_peer))
+ {
+ increment_stats (STAT_HELLOS_PROVIDED);
+ GNUNET_TRANSPORT_offer_hello (transport_handle, hello_msg, NULL, NULL);
+ GNUNET_CORE_peer_request_connect (coreAPI, &new_peer, NULL, NULL);
+ }
+ }
+ }
+
+ record =
+ GNUNET_CONTAINER_multihashmap_get (forward_list.hashmap, &msg_ctx->key);
+
+ if (record == NULL) /* No record of this message! */
+ {
+#if DEBUG_DHT
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "`%s:%s': Have no record of response key %s uid %llu\n",
+ my_short_id, "DHT", GNUNET_h2s (&msg_ctx->key),
+ msg_ctx->unique_id);
+#endif
+ return 0;
+ }
+
+ pos = record->head;
+ while (pos != NULL)
+ {
+ if (0 == memcmp (&pos->source, &my_identity, sizeof (struct
GNUNET_PeerIdentity))) /* Local client (or DHT) initiated request! */
+ {
+#if DEBUG_DHT
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "`%s:%s': Sending response key %s uid %llu to client\n",
+ my_short_id, "DHT", GNUNET_h2s (&msg_ctx->key),
+ msg_ctx->unique_id);
+#endif
+ increment_stats (STAT_RESULTS_TO_CLIENT);
+ if (ntohs (msg->type) == GNUNET_MESSAGE_TYPE_DHT_GET_RESULT)
+ increment_stats (STAT_GET_REPLY);
+#if DEBUG_DHT > 1
+ for (i = 0; i < msg_ctx->path_history_len; i++)
+ {
+ char *path_offset;
+
+ path_offset =
+ &msg_ctx->path_history[i * sizeof (struct GNUNET_PeerIdentity)];
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "(before client) Key %s Found peer %d:%s\n",
+ GNUNET_h2s (&msg_ctx->key), i,
+ GNUNET_i2s ((struct GNUNET_PeerIdentity *) path_offset));
+ }
+#endif
+ send_reply_to_client (pos->client, msg, msg_ctx);
+ }
+ else /* Send to peer */
+ {
+ peer_info = find_peer_by_id (&pos->source);
+ if (peer_info == NULL) /* Didn't find the peer in our routing table,
perhaps peer disconnected! */
+ {
+ pos = pos->next;
+ continue;
+ }
+#if DEBUG_DHT
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "`%s:%s': Forwarding response key %s uid %llu to peer %s\n",
+ my_short_id, "DHT", GNUNET_h2s (&msg_ctx->key),
+ msg_ctx->unique_id, GNUNET_i2s (&peer_info->id));
+#endif
+ forward_result_message (msg, peer_info, msg_ctx);
+ /* Try removing forward entries after sending once, only allows ONE
response per request */
+ if (pos->delete_task != GNUNET_SCHEDULER_NO_TASK)
+ {
+ GNUNET_SCHEDULER_cancel (pos->delete_task);
+ pos->delete_task =
+ GNUNET_SCHEDULER_add_now (&remove_forward_entry, pos);
+ }
+ }
+ pos = pos->next;
+ }
+ return 0;
+}
+
+
+/**
+ * Iterator for local get request results,
+ *
+ * @param cls closure for iterator, a DatacacheGetContext
+ * @param exp when does this value expire?
+ * @param key the key this data is stored under
+ * @param size the size of the data identified by key
+ * @param data the actual data
+ * @param type the type of the data
+ *
+ * @return GNUNET_OK to continue iteration, anything else
+ * to stop iteration.
+ */
+static int
+datacache_get_iterator (void *cls, struct GNUNET_TIME_Absolute exp,
+ const GNUNET_HashCode * key, size_t size,
+ const char *data, enum GNUNET_BLOCK_Type type)
+{
+ struct DHT_MessageContext *msg_ctx = cls;
+ struct DHT_MessageContext new_msg_ctx;
+ struct GNUNET_DHT_GetResultMessage *get_result;
+ enum GNUNET_BLOCK_EvaluationResult eval;
+ const struct DHTPutEntry *put_entry;
+ int get_size;
+ char *path_offset;
+
+#if DEBUG_DHT
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "`%s:%s': Received `%s' response from datacache\n", my_short_id,
+ "DHT", "GET");
+#endif
+
+ put_entry = (const struct DHTPutEntry *) data;
+
+ if (size !=
+ sizeof (struct DHTPutEntry) + put_entry->data_size +
+ (put_entry->path_length * sizeof (struct GNUNET_PeerIdentity)))
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+ "Path + data size doesn't add up for data inserted into
datacache!\nData size %d, path length %d, expected %d, got %d\n",
+ put_entry->data_size, put_entry->path_length,
+ sizeof (struct DHTPutEntry) + put_entry->data_size +
+ (put_entry->path_length * sizeof (struct GNUNET_PeerIdentity)),
+ size);
+ msg_ctx->do_forward = GNUNET_NO;
+ return GNUNET_OK;
+ }
+
+ eval =
+ GNUNET_BLOCK_evaluate (block_context, type, key, &msg_ctx->reply_bf,
+ msg_ctx->reply_bf_mutator, msg_ctx->xquery,
+ msg_ctx->xquery_size, &put_entry[1],
+ put_entry->data_size);
+
+ switch (eval)
+ {
+ case GNUNET_BLOCK_EVALUATION_OK_LAST:
+ msg_ctx->do_forward = GNUNET_NO;
+ case GNUNET_BLOCK_EVALUATION_OK_MORE:
+ memcpy (&new_msg_ctx, msg_ctx, sizeof (struct DHT_MessageContext));
+ if (GNUNET_DHT_RO_RECORD_ROUTE ==
+ (msg_ctx->msg_options & GNUNET_DHT_RO_RECORD_ROUTE))
+ {
+ new_msg_ctx.msg_options = GNUNET_DHT_RO_RECORD_ROUTE;
+ }
+
+ get_size =
+ sizeof (struct GNUNET_DHT_GetResultMessage) + put_entry->data_size +
+ (put_entry->path_length * sizeof (struct GNUNET_PeerIdentity));
+ get_result = GNUNET_malloc (get_size);
+ get_result->header.type = htons (GNUNET_MESSAGE_TYPE_DHT_GET_RESULT);
+ get_result->header.size = htons (get_size);
+ get_result->expiration = GNUNET_TIME_absolute_hton (exp);
+ get_result->type = htons (type);
+ get_result->put_path_length = htons (put_entry->path_length);
+ path_offset = (char *) &put_entry[1];
+ path_offset += put_entry->data_size;
+ /* Copy the actual data and the path_history to the end of the get result
*/
+ memcpy (&get_result[1], &put_entry[1],
+ put_entry->data_size +
+ (put_entry->path_length * sizeof (struct GNUNET_PeerIdentity)));
+ new_msg_ctx.peer = my_identity;
+ new_msg_ctx.bloom = NULL;
+ new_msg_ctx.hop_count = 0;
+ new_msg_ctx.importance = DHT_DEFAULT_P2P_IMPORTANCE + 2; /* Make result
routing a higher priority */
+ new_msg_ctx.timeout = DHT_DEFAULT_P2P_TIMEOUT;
+ increment_stats (STAT_GET_RESPONSE_START);
+ route_result_message (&get_result->header, &new_msg_ctx);
+ GNUNET_free (get_result);
+ break;
+ case GNUNET_BLOCK_EVALUATION_OK_DUPLICATE:
+#if DEBUG_DHT
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "`%s:%s': Duplicate block error\n",
+ my_short_id, "DHT");
+#endif
+ break;
+ case GNUNET_BLOCK_EVALUATION_RESULT_INVALID:
+#if DEBUG_DHT
+ GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "`%s:%s': Invalid request error\n",
+ my_short_id, "DHT");
+#endif
+ break;
+ case GNUNET_BLOCK_EVALUATION_REQUEST_VALID:
+#if DEBUG_DHT
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "`%s:%s': Valid request, no results.\n", my_short_id, "DHT");
+#endif
+ GNUNET_break (0);
+ break;
+ case GNUNET_BLOCK_EVALUATION_REQUEST_INVALID:
+ GNUNET_break_op (0);
+ msg_ctx->do_forward = GNUNET_NO;
+ break;
+ case GNUNET_BLOCK_EVALUATION_TYPE_NOT_SUPPORTED:
+#if DEBUG_DHT
+ GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+ "`%s:%s': Unsupported block type (%u) in response!\n",
+ my_short_id, "DHT", type);
+#endif
+ /* msg_ctx->do_forward = GNUNET_NO; // not sure... */
+ break;
+ }
+ return GNUNET_OK;
+}
+
+
+/**
+ * Main function that handles whether or not to route a message to other
+ * peers.
+ *
+ * @param msg the message to be routed
+ * @param msg_ctx the context containing all pertinent information about the
message
+ */
+static void
+route_message (const struct GNUNET_MessageHeader *msg,
+ struct DHT_MessageContext *msg_ctx);
+
+
+/**
+ * Server handler for all dht get requests, look for data,
+ * if found, send response either to clients or other peers.
+ *
+ * @param msg the actual get message
+ * @param msg_ctx struct containing pertinent information about the get request
+ *
+ * @return number of items found for GET request
+ */
+static unsigned int
+handle_dht_get (const struct GNUNET_MessageHeader *msg,
+ struct DHT_MessageContext *msg_ctx)
+{
+ const struct GNUNET_DHT_GetMessage *get_msg;
+ uint16_t msize;
+ uint16_t bf_size;
+ unsigned int results;
+ const char *end;
+ enum GNUNET_BLOCK_Type type;
+
+ msize = ntohs (msg->size);
+ if (msize < sizeof (struct GNUNET_DHT_GetMessage))
+ {
+ GNUNET_break (0);
+ return 0;
+ }
+ get_msg = (const struct GNUNET_DHT_GetMessage *) msg;
+ bf_size = ntohs (get_msg->bf_size);
+ msg_ctx->xquery_size = ntohs (get_msg->xquery_size);
+ msg_ctx->reply_bf_mutator = get_msg->bf_mutator;
+ if (msize !=
+ sizeof (struct GNUNET_DHT_GetMessage) + bf_size + msg_ctx->xquery_size)
+ {
+ GNUNET_break_op (0);
+ return 0;
+ }
+ end = (const char *) &get_msg[1];
+ if (msg_ctx->xquery_size == 0)
+ {
+ msg_ctx->xquery = NULL;
+ }
+ else
+ {
+ msg_ctx->xquery = (const void *) end;
+ end += msg_ctx->xquery_size;
+ }
+ if (bf_size == 0)
+ {
+ msg_ctx->reply_bf = NULL;
+ }
+ else
+ {
+ msg_ctx->reply_bf =
+ GNUNET_CONTAINER_bloomfilter_init (end, bf_size,
+ GNUNET_DHT_GET_BLOOMFILTER_K);
+ }
+ type = (enum GNUNET_BLOCK_Type) ntohl (get_msg->type);
+#if DEBUG_DHT
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "`%s:%s': Received `%s' request, message type %u, key %s, uid
%llu\n",
+ my_short_id, "DHT", "GET", type, GNUNET_h2s (&msg_ctx->key),
+ msg_ctx->unique_id);
+#endif
+ increment_stats (STAT_GETS);
+ results = 0;
+ msg_ctx->do_forward = GNUNET_YES;
+ if (datacache != NULL)
+ results =
+ GNUNET_DATACACHE_get (datacache, &msg_ctx->key, type,
+ &datacache_get_iterator, msg_ctx);
+#if DEBUG_DHT
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "`%s:%s': Found %d results for `%s' request uid %llu\n",
+ my_short_id, "DHT", results, "GET", msg_ctx->unique_id);
+#endif
+ if (results >= 1)
+ {
+ }
+ else
+ {
+ /* check query valid */
+ if (GNUNET_BLOCK_EVALUATION_REQUEST_INVALID ==
+ GNUNET_BLOCK_evaluate (block_context, type, &msg_ctx->key,
+ &msg_ctx->reply_bf, msg_ctx->reply_bf_mutator,
+ msg_ctx->xquery, msg_ctx->xquery_size, NULL, 0))
+ {
+ GNUNET_break_op (0);
+ msg_ctx->do_forward = GNUNET_NO;
+ }
+ }
+
+ if (msg_ctx->do_forward == GNUNET_YES)
+ route_message (msg, msg_ctx);
+ GNUNET_CONTAINER_bloomfilter_free (msg_ctx->reply_bf);
+ return results;
+}
+
+
+static void
+remove_recent_find_peer (void *cls,
+ const struct GNUNET_SCHEDULER_TaskContext *tc)
+{
+ GNUNET_HashCode *key = cls;
+
+ GNUNET_assert (GNUNET_YES ==
+ GNUNET_CONTAINER_multihashmap_remove
+ (recent_find_peer_requests, key, NULL));
+ GNUNET_free (key);
+}
+
+
+/**
+ * Server handler for initiating local dht find peer requests
+ *
+ * @param find_msg the actual find peer message
+ * @param msg_ctx struct containing pertinent information about the request
+ *
+ */
+static void
+handle_dht_find_peer (const struct GNUNET_MessageHeader *find_msg,
+ struct DHT_MessageContext *msg_ctx)
+{
+ struct GNUNET_MessageHeader *find_peer_result;
+ struct GNUNET_DHT_FindPeerMessage *find_peer_message;
+ struct DHT_MessageContext *new_msg_ctx;
+ struct GNUNET_CONTAINER_BloomFilter *incoming_bloom;
+ size_t hello_size;
+ size_t tsize;
+ GNUNET_HashCode *recent_hash;
+ struct GNUNET_MessageHeader *other_hello;
+ size_t other_hello_size;
+ struct GNUNET_PeerIdentity peer_id;
+
+ find_peer_message = (struct GNUNET_DHT_FindPeerMessage *) find_msg;
+ GNUNET_break_op (ntohs (find_msg->size) >=
+ (sizeof (struct GNUNET_DHT_FindPeerMessage)));
+ if (ntohs (find_msg->size) < sizeof (struct GNUNET_DHT_FindPeerMessage))
+ return;
+ other_hello = NULL;
+ other_hello_size = 0;
+ if (ntohs (find_msg->size) > sizeof (struct GNUNET_DHT_FindPeerMessage))
+ {
+ other_hello_size =
+ ntohs (find_msg->size) - sizeof (struct GNUNET_DHT_FindPeerMessage);
+ other_hello = GNUNET_malloc (other_hello_size);
+ memcpy (other_hello, &find_peer_message[1], other_hello_size);
+ if ((GNUNET_HELLO_size ((struct GNUNET_HELLO_Message *) other_hello) == 0)
+ || (GNUNET_OK !=
+ GNUNET_HELLO_get_id ((struct GNUNET_HELLO_Message *) other_hello,
+ &peer_id)))
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+ "Received invalid HELLO message in find peer request!\n");
+ GNUNET_free (other_hello);
+ return;
+ }
+#if FIND_PEER_WITH_HELLO
+ if (GNUNET_YES == consider_peer (&peer_id))
+ {
+ increment_stats (STAT_HELLOS_PROVIDED);
+ GNUNET_TRANSPORT_offer_hello (transport_handle, other_hello, NULL, NULL);
+ GNUNET_CORE_peer_request_connect (coreAPI, &peer_id, NULL, NULL);
+ route_message (find_msg, msg_ctx);
+ GNUNET_free (other_hello);
+ return;
+ }
+ else /* We don't want this peer! */
+ {
+ route_message (find_msg, msg_ctx);
+ GNUNET_free (other_hello);
+ return;
+ }
+#endif
+ }
+
+#if DEBUG_DHT
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "`%s:%s': Received `%s' request from client, key %s (msg size
%d, we expected %d)\n",
+ my_short_id, "DHT", "FIND PEER", GNUNET_h2s (&msg_ctx->key),
+ ntohs (find_msg->size), sizeof (struct GNUNET_MessageHeader));
+#endif
+ if (my_hello == NULL)
+ {
+#if DEBUG_DHT
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "`%s': Our HELLO is null, can't return.\n", "DHT");
+#endif
+ GNUNET_free_non_null (other_hello);
+ route_message (find_msg, msg_ctx);
+ return;
+ }
+
+ incoming_bloom =
+ GNUNET_CONTAINER_bloomfilter_init (find_peer_message->bloomfilter,
+ DHT_BLOOM_SIZE, DHT_BLOOM_K);
+ if (GNUNET_YES ==
+ GNUNET_CONTAINER_bloomfilter_test (incoming_bloom,
+ &my_identity.hashPubKey))
+ {
+ increment_stats (STAT_BLOOM_FIND_PEER);
+ GNUNET_CONTAINER_bloomfilter_free (incoming_bloom);
+ GNUNET_free_non_null (other_hello);
+ route_message (find_msg, msg_ctx);
+ return; /* We match the bloomfilter, do not send a
response to this peer (they likely already know us!) */
+ }
+ GNUNET_CONTAINER_bloomfilter_free (incoming_bloom);
+
+ /**
+ * Ignore any find peer requests from a peer we have seen very recently.
+ */
+ if (GNUNET_YES == GNUNET_CONTAINER_multihashmap_contains
(recent_find_peer_requests, &msg_ctx->key)) /* We have recently responded to a
find peer request for this peer! */
+ {
+ increment_stats ("# dht find peer requests ignored (recently seen!)");
+ GNUNET_free_non_null (other_hello);
+ return;
+ }
+
+ /**
+ * Use this check to only allow the peer to respond to find peer requests if
+ * it would be beneficial to have the requesting peer in this peers routing
+ * table. Can be used to thwart peers flooding the network with find peer
+ * requests that we don't care about. However, if a new peer is joining
+ * the network and has no other peers this is a problem (assume all buckets
+ * full, no one will respond!).
+ */
+ memcpy (&peer_id.hashPubKey, &msg_ctx->key, sizeof (GNUNET_HashCode));
+ if (GNUNET_NO == consider_peer (&peer_id))
+ {
+ increment_stats ("# dht find peer requests ignored (do not need!)");
+ GNUNET_free_non_null (other_hello);
+ route_message (find_msg, msg_ctx);
+ return;
+ }
+
+ recent_hash = GNUNET_malloc (sizeof (GNUNET_HashCode));
+ memcpy (recent_hash, &msg_ctx->key, sizeof (GNUNET_HashCode));
+ if (GNUNET_SYSERR !=
+ GNUNET_CONTAINER_multihashmap_put (recent_find_peer_requests,
+ &msg_ctx->key, NULL,
+
GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY))
+ {
+#if DEBUG_DHT
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Adding recent remove task for key `%s`!\n",
+ GNUNET_h2s (&msg_ctx->key));
+#endif
+ /* Only add a task if there wasn't one for this key already! */
+ GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_relative_multiply
+ (GNUNET_TIME_UNIT_SECONDS, 30),
+ &remove_recent_find_peer, recent_hash);
+ }
+ else
+ {
+ GNUNET_free (recent_hash);
+#if DEBUG_DHT
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Received duplicate find peer request too soon!\n");
+#endif
+ }
+
+ /* Simplistic find_peer functionality, always return our hello */
+ hello_size = ntohs (my_hello->size);
+ tsize = hello_size + sizeof (struct GNUNET_MessageHeader);
+
+ if (tsize >= GNUNET_SERVER_MAX_MESSAGE_SIZE)
+ {
+ GNUNET_break_op (0);
+ GNUNET_free_non_null (other_hello);
+ return;
+ }
+
+ find_peer_result = GNUNET_malloc (tsize);
+ find_peer_result->type = htons (GNUNET_MESSAGE_TYPE_DHT_FIND_PEER_RESULT);
+ find_peer_result->size = htons (tsize);
+ memcpy (&find_peer_result[1], my_hello, hello_size);
+#if DEBUG_DHT
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "`%s': Sending hello size %d to requesting peer.\n", "DHT",
+ hello_size);
+#endif
+
+ new_msg_ctx = GNUNET_malloc (sizeof (struct DHT_MessageContext));
+ memcpy (new_msg_ctx, msg_ctx, sizeof (struct DHT_MessageContext));
+ new_msg_ctx->peer = my_identity;
+ new_msg_ctx->bloom =
+ GNUNET_CONTAINER_bloomfilter_init (NULL, DHT_BLOOM_SIZE, DHT_BLOOM_K);
+ new_msg_ctx->hop_count = 0;
+ new_msg_ctx->importance = DHT_DEFAULT_P2P_IMPORTANCE + 2; /* Make find
peer requests a higher priority */
+ new_msg_ctx->timeout = DHT_DEFAULT_P2P_TIMEOUT;
+ increment_stats (STAT_FIND_PEER_ANSWER);
+ if (GNUNET_DHT_RO_RECORD_ROUTE ==
+ (msg_ctx->msg_options & GNUNET_DHT_RO_RECORD_ROUTE))
+ {
+ new_msg_ctx->msg_options = GNUNET_DHT_RO_RECORD_ROUTE;
+ new_msg_ctx->path_history_len = msg_ctx->path_history_len;
+ /* Assign to previous msg_ctx path history, caller should free after our
return */
+ new_msg_ctx->path_history = msg_ctx->path_history;
+ }
+ route_result_message (find_peer_result, new_msg_ctx);
+ GNUNET_free (new_msg_ctx);
+ GNUNET_free_non_null (other_hello);
+ GNUNET_free (find_peer_result);
+ route_message (find_msg, msg_ctx);
+}
+
+
+/**
+ * Server handler for initiating local dht put requests
+ *
+ * @param msg the actual put message
+ * @param msg_ctx struct containing pertinent information about the request
+ */
+static void
+handle_dht_put (const struct GNUNET_MessageHeader *msg,
+ struct DHT_MessageContext *msg_ctx)
+{
+ const struct GNUNET_DHT_PutMessage *put_msg;
+ struct DHTPutEntry *put_entry;
+ unsigned int put_size;
+ char *path_offset;
+ enum GNUNET_BLOCK_Type put_type;
+ size_t data_size;
+ int ret;
+ GNUNET_HashCode key;
+ struct DHTQueryRecord *record;
+
+ GNUNET_assert (ntohs (msg->size) >= sizeof (struct GNUNET_DHT_PutMessage));
+
+ put_msg = (const struct GNUNET_DHT_PutMessage *) msg;
+ put_type = (enum GNUNET_BLOCK_Type) ntohl (put_msg->type);
+ data_size =
+ ntohs (put_msg->header.size) - sizeof (struct GNUNET_DHT_PutMessage);
+ ret =
+ GNUNET_BLOCK_get_key (block_context, put_type, &put_msg[1], data_size,
+ &key);
+ if (GNUNET_NO == ret)
+ {
+ /* invalid reply */
+ GNUNET_break_op (0);
+ return;
+ }
+ if ((GNUNET_YES == ret) &&
+ (0 != memcmp (&key, &msg_ctx->key, sizeof (GNUNET_HashCode))))
+ {
+ /* invalid wrapper: key mismatch! */
+ GNUNET_break_op (0);
+ return;
+ }
+ /* ret == GNUNET_SYSERR means that there is no known relationship between
+ * data and the key, so we cannot check it */
+#if DEBUG_DHT
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "`%s:%s': Received `%s' request (inserting data!), message type
%d, key %s, uid %llu\n",
+ my_short_id, "DHT", "PUT", put_type, GNUNET_h2s (&msg_ctx->key),
+ msg_ctx->unique_id);
+#endif
+
+ record = GNUNET_CONTAINER_multihashmap_get(forward_list.hashmap,
+ &msg_ctx->key);
+ if (NULL != record)
+ {
+ struct DHTRouteSource *pos;
+ struct GNUNET_DHT_GetResultMessage *get_result;
+ struct DHT_MessageContext new_msg_ctx;
+ size_t get_size;
+
+ pos = record->head;
+ while (pos != NULL)
+ {
+ /* TODO: do only for local started requests? or also for remote peers? */
+ /* TODO: include this in statistics? under what? */
+ /* TODO: reverse order of path_history? */
+ if (NULL == pos->client)
+ {
+ pos = pos->next;
+ continue;
+ }
+
+ memcpy (&new_msg_ctx, msg_ctx, sizeof (struct DHT_MessageContext));
+ if (GNUNET_DHT_RO_RECORD_ROUTE ==
+ (msg_ctx->msg_options & GNUNET_DHT_RO_RECORD_ROUTE))
+ {
+ new_msg_ctx.msg_options = GNUNET_DHT_RO_RECORD_ROUTE;
+ }
+
+ get_size =
+ sizeof (struct GNUNET_DHT_GetResultMessage) + data_size +
+ (msg_ctx->path_history_len * sizeof (struct GNUNET_PeerIdentity));
+ get_result = GNUNET_malloc (get_size);
+ get_result->header.type = htons (GNUNET_MESSAGE_TYPE_DHT_GET_RESULT);
+ get_result->header.size = htons (get_size);
+ get_result->expiration = put_msg->expiration;
+ get_result->type = put_msg->type;
+ get_result->put_path_length = htons (msg_ctx->path_history_len);
+
+ /* Copy the actual data and the path_history to the end of the get
result */
+ memcpy (&get_result[1], &put_msg[1], data_size);
+ path_offset = (char *) &get_result[1];
+ path_offset += data_size;
+ memcpy (path_offset, msg_ctx->path_history,
+ msg_ctx->path_history_len * sizeof (struct GNUNET_PeerIdentity));
+ new_msg_ctx.peer = my_identity;
+ new_msg_ctx.bloom = NULL;
+ new_msg_ctx.hop_count = 0;
+ /* Make result routing a higher priority */
+ new_msg_ctx.importance = DHT_DEFAULT_P2P_IMPORTANCE + 2;
+ new_msg_ctx.timeout = DHT_DEFAULT_P2P_TIMEOUT;
+ new_msg_ctx.unique_id = pos->uid;
+ send_reply_to_client(pos->client, &get_result->header, &new_msg_ctx);
+ GNUNET_free (get_result);
+ pos = pos->next;
+ }
+ }
+
+ if (msg_ctx->closest != GNUNET_YES)
+ {
+ route_message (msg, msg_ctx);
+ return;
+ }
+
+#if DEBUG_DHT
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "`%s:%s': Received `%s' request (inserting data!), message type
%d, key %s, uid %llu\n",
+ my_short_id, "DHT", "PUT", put_type, GNUNET_h2s (&msg_ctx->key),
+ msg_ctx->unique_id);
+#endif
+
+ increment_stats (STAT_PUTS_INSERTED);
+ if (datacache != NULL)
+ {
+ /* Put size is actual data size plus struct overhead plus path length (if
any) */
+ put_size =
+ data_size + sizeof (struct DHTPutEntry) +
+ (msg_ctx->path_history_len * sizeof (struct GNUNET_PeerIdentity));
+ put_entry = GNUNET_malloc (put_size);
+ put_entry->data_size = data_size;
+ put_entry->path_length = msg_ctx->path_history_len;
+ /* Copy data to end of put entry */
+ memcpy (&put_entry[1], &put_msg[1], data_size);
+ if (msg_ctx->path_history_len > 0)
+ {
+ /* Copy path after data */
+ path_offset = (char *) &put_entry[1];
+ path_offset += data_size;
+ memcpy (path_offset, msg_ctx->path_history,
+ msg_ctx->path_history_len * sizeof (struct GNUNET_PeerIdentity));
+ }
+
+ ret =
+ GNUNET_DATACACHE_put (datacache, &msg_ctx->key, put_size,
+ (const char *) put_entry, put_type,
+ GNUNET_TIME_absolute_ntoh (put_msg->expiration));
+ GNUNET_free (put_entry);
+ }
+ else
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "`%s:%s': %s request received, but have no datacache!\n",
+ my_short_id, "DHT", "PUT");
+
+ route_message (msg, msg_ctx);
+}
+
+
+/**
+ * To how many peers should we (on average)
+ * forward the request to obtain the desired
+ * target_replication count (on average).
+ *
+ * returns: target_replication / (est. hops) + (target_replication * hop_count)
+ * where est. hops is typically 2 * the routing table depth
+ *
+ * @param hop_count number of hops the message has traversed
+ * @param target_replication the number of total paths desired
+ *
+ * @return Some number of peers to forward the message to
+ */
+static unsigned int
+get_forward_count (unsigned int hop_count, size_t target_replication)
+{
+ uint32_t random_value;
+ unsigned int forward_count;
+ float target_value;
+
+ if (hop_count > log_of_network_size_estimate * 4.0)
+ {
+ /* forcefully terminate */
+ return 0;
+ }
+
+ if (hop_count > log_of_network_size_estimate * 2.0)
+ {
+ /* keep forwarding, but no more replication */
+ return 1;
+ }
+
+ target_value =
+ 1 + (target_replication - 1.0) / (log_of_network_size_estimate +
+ ((float) (target_replication - 1.0) *
+ hop_count));
+ /* Set forward count to floor of target_value */
+ forward_count = (unsigned int) target_value;
+ /* Subtract forward_count (floor) from target_value (yields value between 0
and 1) */
+ target_value = target_value - forward_count;
+ random_value =
+ GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_STRONG, UINT32_MAX);
+ if (random_value < (target_value * UINT32_MAX))
+ forward_count++;
+ return forward_count;
+}
+
+
+/**
+ * Check whether my identity is closer than any known peers.
+ * If a non-null bloomfilter is given, check if this is the closest
+ * peer that hasn't already been routed to.
+ *
+ * @param target hash code to check closeness to
+ * @param bloom bloomfilter, exclude these entries from the decision
+ * @return GNUNET_YES if node location is closest,
+ * GNUNET_NO otherwise.
+ */
+static int
+am_closest_peer (const GNUNET_HashCode * target,
+ struct GNUNET_CONTAINER_BloomFilter *bloom)
+{
+ int bits;
+ int other_bits;
+ int bucket_num;
+ int count;
+ struct PeerInfo *pos;
+ unsigned int my_distance;
+
+ if (0 == memcmp (&my_identity.hashPubKey, target, sizeof (GNUNET_HashCode)))
+ return GNUNET_YES;
+
+ bucket_num = find_current_bucket (target);
+
+ bits = GNUNET_CRYPTO_hash_matching_bits (&my_identity.hashPubKey, target);
+ my_distance = distance (&my_identity.hashPubKey, target);
+ pos = k_buckets[bucket_num].head;
+ count = 0;
+ while ((pos != NULL) && (count < bucket_size))
+ {
+ if ((bloom != NULL) &&
+ (GNUNET_YES ==
+ GNUNET_CONTAINER_bloomfilter_test (bloom, &pos->id.hashPubKey)))
+ {
+ pos = pos->next;
+ continue; /* Skip already checked entries */
+ }
+
+ other_bits = GNUNET_CRYPTO_hash_matching_bits (&pos->id.hashPubKey,
target);
+ if (other_bits > bits)
+ return GNUNET_NO;
+ else if (other_bits == bits) /* We match the same number of bits */
+ {
+ if (distance (&pos->id.hashPubKey, target) < my_distance) /* Check all
known peers, only return if we are the true closest */
+ return GNUNET_NO;
+ }
+ pos = pos->next;
+ }
+
+ /* No peers closer, we are the closest! */
+ return GNUNET_YES;
+}
+
+
+/**
+ * Select a peer from the routing table that would be a good routing
+ * destination for sending a message for "target". The resulting peer
+ * must not be in the set of blocked peers.<p>
+ *
+ * Note that we should not ALWAYS select the closest peer to the
+ * target, peers further away from the target should be chosen with
+ * exponentially declining probability.
+ *
+ * @param target the key we are selecting a peer to route to
+ * @param bloom a bloomfilter containing entries this request has seen already
+ * @param hops how many hops has this message traversed thus far
+ *
+ * @return Peer to route to, or NULL on error
+ */
+static struct PeerInfo *
+select_peer (const GNUNET_HashCode * target,
+ struct GNUNET_CONTAINER_BloomFilter *bloom, unsigned int hops)
+{
+ unsigned int bc;
+ unsigned int count;
+ unsigned int selected;
+ struct PeerInfo *pos;
+ unsigned int distance;
+ unsigned int largest_distance;
+ struct PeerInfo *chosen;
+
+ if (hops >= log_of_network_size_estimate)
+ {
+ /* greedy selection (closest peer that is not in bloomfilter) */
+ largest_distance = 0;
+ chosen = NULL;
+ for (bc = lowest_bucket; bc < MAX_BUCKETS; bc++)
+ {
+ pos = k_buckets[bc].head;
+ count = 0;
+ while ((pos != NULL) && (count < bucket_size))
+ {
+ /* If we are doing strict Kademlia routing, then checking the
bloomfilter is basically cheating! */
+ if (GNUNET_NO ==
+ GNUNET_CONTAINER_bloomfilter_test (bloom, &pos->id.hashPubKey))
+ {
+ distance = inverse_distance (target, &pos->id.hashPubKey);
+ if (distance > largest_distance)
+ {
+ chosen = pos;
+ largest_distance = distance;
+ }
+ }
+ count++;
+ pos = pos->next;
+ }
+ }
+ if ((largest_distance > 0) && (chosen != NULL))
+ {
+ GNUNET_CONTAINER_bloomfilter_add (bloom, &chosen->id.hashPubKey);
+ return chosen;
+ }
+ return NULL; /* no peer available or we are the closest */
+ }
+
+
+ /* select "random" peer */
+ /* count number of peers that are available and not filtered */
+ count = 0;
+ for (bc = lowest_bucket; bc < MAX_BUCKETS; bc++)
+ {
+ pos = k_buckets[bc].head;
+ while ((pos != NULL) && (count < bucket_size))
+ {
+ if (GNUNET_YES ==
+ GNUNET_CONTAINER_bloomfilter_test (bloom, &pos->id.hashPubKey))
+ {
+ pos = pos->next;
+ increment_stats ("# peer blocked from selection by Bloom filter");
+ continue; /* Ignore bloomfiltered peers */
+ }
+ count++;
+ pos = pos->next;
+ }
+ }
+ if (count == 0) /* No peers to select from! */
+ {
+ increment_stats ("# failed to select peer");
+ return NULL;
+ }
+ /* Now actually choose a peer */
+ selected = GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK, count);
+ count = 0;
+ for (bc = lowest_bucket; bc < MAX_BUCKETS; bc++)
+ {
+ pos = k_buckets[bc].head;
+ while ((pos != NULL) && (count < bucket_size))
+ {
+ if (GNUNET_YES ==
+ GNUNET_CONTAINER_bloomfilter_test (bloom, &pos->id.hashPubKey))
+ {
+ pos = pos->next;
+ continue; /* Ignore bloomfiltered peers */
+ }
+ if (0 == selected--)
+ return pos;
+ pos = pos->next;
+ }
+ }
+ GNUNET_break (0);
+ return NULL;
+}
+
+
+/**
+ * Remember this routing request so that if a reply is
+ * received we can either forward it to the correct peer
+ * or return the result locally.
+ *
+ * @param msg_ctx Context of the route request
+ *
+ * @return GNUNET_YES if this response was cached, GNUNET_NO if not
+ */
+static int
+cache_response (struct DHT_MessageContext *msg_ctx)
+{
+ struct DHTQueryRecord *record;
+ struct DHTRouteSource *source_info;
+ struct DHTRouteSource *pos;
+ struct GNUNET_TIME_Absolute now;
+ unsigned int current_size;
+
+ current_size = GNUNET_CONTAINER_multihashmap_size (forward_list.hashmap);
+
+ while (current_size >= MAX_OUTSTANDING_FORWARDS)
+ {
+ source_info = GNUNET_CONTAINER_heap_remove_root (forward_list.minHeap);
+ GNUNET_assert (source_info != NULL);
+ record = source_info->record;
+ GNUNET_CONTAINER_DLL_remove (record->head, record->tail, source_info);
+ if (record->head == NULL) /* No more entries in DLL */
+ {
+ GNUNET_assert (GNUNET_YES ==
+ GNUNET_CONTAINER_multihashmap_remove
(forward_list.hashmap,
+ &record->key,
+ record));
+ GNUNET_free (record);
+ }
+ if (source_info->delete_task != GNUNET_SCHEDULER_NO_TASK)
+ {
+ GNUNET_SCHEDULER_cancel (source_info->delete_task);
+ source_info->delete_task = GNUNET_SCHEDULER_NO_TASK;
+ }
+ if (source_info->find_peers_responded != NULL)
+ GNUNET_CONTAINER_bloomfilter_free (source_info->find_peers_responded);
+ GNUNET_free (source_info);
+ current_size = GNUNET_CONTAINER_multihashmap_size (forward_list.hashmap);
+ }
+
+ /** Non-local request and have too many outstanding forwards, discard! */
+ if ((current_size >= MAX_OUTSTANDING_FORWARDS) && (msg_ctx->client == NULL))
+ return GNUNET_NO;
+
+ now = GNUNET_TIME_absolute_get ();
+ record =
+ GNUNET_CONTAINER_multihashmap_get (forward_list.hashmap, &msg_ctx->key);
+ if (record != NULL) /* Already know this request! */
+ {
+ pos = record->head;
+ while (pos != NULL)
+ {
+ if (0 ==
+ memcmp (&msg_ctx->peer, &pos->source,
+ sizeof (struct GNUNET_PeerIdentity)))
+ break; /* Already have this peer in reply list! */
+ pos = pos->next;
+ }
+ if ((pos != NULL) && (pos->client == msg_ctx->client)) /* Seen this
already */
+ {
+ GNUNET_CONTAINER_heap_update_cost (forward_list.minHeap, pos->hnode,
+ now.abs_value);
+ return GNUNET_NO;
+ }
+ }
+ else
+ {
+ record = GNUNET_malloc (sizeof (struct DHTQueryRecord));
+ GNUNET_assert (GNUNET_OK ==
+ GNUNET_CONTAINER_multihashmap_put (forward_list.hashmap,
+ &msg_ctx->key, record,
+
GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
+ memcpy (&record->key, &msg_ctx->key, sizeof (GNUNET_HashCode));
+ }
+
+ source_info = GNUNET_malloc (sizeof (struct DHTRouteSource));
+ source_info->record = record;
+ source_info->delete_task =
+ GNUNET_SCHEDULER_add_delayed (DHT_FORWARD_TIMEOUT, &remove_forward_entry,
+ source_info);
+ source_info->find_peers_responded =
+ GNUNET_CONTAINER_bloomfilter_init (NULL, DHT_BLOOM_SIZE, DHT_BLOOM_K);
+ source_info->source = msg_ctx->peer;
+ GNUNET_CONTAINER_DLL_insert_after (record->head, record->tail, record->tail,
+ source_info);
+ if (msg_ctx->client != NULL) /* For local request, set timeout so high it
effectively never gets pushed out */
+ {
+ source_info->client = msg_ctx->client;
+ now = GNUNET_TIME_absolute_get_forever ();
+ }
+ source_info->hnode =
+ GNUNET_CONTAINER_heap_insert (forward_list.minHeap, source_info,
+ now.abs_value);
+ source_info->uid = msg_ctx->unique_id;
+#if DEBUG_DHT > 1
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "`%s:%s': Created new forward source info for %s uid %llu\n",
+ my_short_id, "DHT", GNUNET_h2s (&msg_ctx->key),
+ msg_ctx->unique_id);
+#endif
+ return GNUNET_YES;
+}
+
+
+/**
+ * Main function that handles whether or not to route a message to other
+ * peers.
+ *
+ * @param msg the message to be routed
+ * @param msg_ctx the context containing all pertinent information about the
message
+ */
+static void
+route_message (const struct GNUNET_MessageHeader *msg,
+ struct DHT_MessageContext *msg_ctx)
+{
+ int i;
+ struct PeerInfo *selected;
+ unsigned int target_forward_count;
+ unsigned int forward_count;
+ struct RecentRequest *recent_req;
+ char *stat_forward_count;
+ char *temp_stat_str;
+
+ increment_stats (STAT_ROUTES);
+ target_forward_count =
+ get_forward_count (msg_ctx->hop_count, msg_ctx->replication);
+ GNUNET_asprintf (&stat_forward_count, "# forward counts of %d",
+ target_forward_count);
+ increment_stats (stat_forward_count);
+ GNUNET_free (stat_forward_count);
+ if (msg_ctx->bloom == NULL)
+ msg_ctx->bloom =
+ GNUNET_CONTAINER_bloomfilter_init (NULL, DHT_BLOOM_SIZE, DHT_BLOOM_K);
+
+ if (GNUNET_CONTAINER_heap_get_size (recent_heap) >= DHT_MAX_RECENT)
+ {
+ recent_req = GNUNET_CONTAINER_heap_peek (recent_heap);
+ GNUNET_assert (recent_req != NULL);
+ GNUNET_SCHEDULER_cancel (recent_req->remove_task);
+ GNUNET_CONTAINER_heap_remove_node (recent_req->heap_node);
+ GNUNET_CONTAINER_bloomfilter_free (recent_req->bloom);
+ GNUNET_free (recent_req);
+ }
+
+ recent_req = GNUNET_malloc (sizeof (struct RecentRequest));
+ recent_req->uid = msg_ctx->unique_id;
+ memcpy (&recent_req->key, &msg_ctx->key, sizeof (GNUNET_HashCode));
+ recent_req->heap_node =
+ GNUNET_CONTAINER_heap_insert (recent_heap, recent_req,
+ GNUNET_TIME_absolute_get ().abs_value);
+ recent_req->bloom =
+ GNUNET_CONTAINER_bloomfilter_init (NULL, DHT_BLOOM_SIZE, DHT_BLOOM_K);
+
+ forward_count = 0;
+ for (i = 0; i < target_forward_count; i++)
+ {
+ selected = select_peer (&msg_ctx->key, msg_ctx->bloom, msg_ctx->hop_count);
+ if (selected == NULL)
+ break;
+ forward_count++;
+ if (GNUNET_CRYPTO_hash_matching_bits
+ (&selected->id.hashPubKey,
+ &msg_ctx->key) >=
+ GNUNET_CRYPTO_hash_matching_bits (&my_identity.hashPubKey,
+ &msg_ctx->key))
+ GNUNET_asprintf (&temp_stat_str,
+ "# requests routed to close(r) peer hop %u",
+ msg_ctx->hop_count);
+ else
+ GNUNET_asprintf (&temp_stat_str,
+ "# requests routed to less close peer hop %u",
+ msg_ctx->hop_count);
+ if (temp_stat_str != NULL)
+ {
+ increment_stats (temp_stat_str);
+ GNUNET_free (temp_stat_str);
+ }
+ GNUNET_CONTAINER_bloomfilter_add (msg_ctx->bloom,
+ &selected->id.hashPubKey);
+ forward_message (msg, selected, msg_ctx);
+ }
+
+ if (msg_ctx->bloom != NULL)
+ {
+ GNUNET_CONTAINER_bloomfilter_or2 (recent_req->bloom, msg_ctx->bloom,
+ DHT_BLOOM_SIZE);
+ GNUNET_CONTAINER_bloomfilter_free (msg_ctx->bloom);
+ msg_ctx->bloom = NULL;
+ }
+}
+
+
+/**
+ * Main function that handles whether or not to route a message to other
+ * peers.
+ *
+ * @param msg the message to be routed
+ * @param msg_ctx the context containing all pertinent information about the
message
+ */
+static void
+demultiplex_message (const struct GNUNET_MessageHeader *msg,
+ struct DHT_MessageContext *msg_ctx)
+{
+ /* FIXME: Should we use closest excluding those we won't route to (the
bloomfilter problem)? */
+ msg_ctx->closest = am_closest_peer (&msg_ctx->key, msg_ctx->bloom);
+
+ switch (ntohs (msg->type))
+ {
+ case GNUNET_MESSAGE_TYPE_DHT_GET: /* Add to hashmap of requests seen,
search for data (always) */
+ cache_response (msg_ctx);
+ handle_dht_get (msg, msg_ctx);
+ break;
+ case GNUNET_MESSAGE_TYPE_DHT_PUT: /* Check if closest, if so insert data.
*/
+ increment_stats (STAT_PUTS);
+ handle_dht_put (msg, msg_ctx);
+ break;
+ case GNUNET_MESSAGE_TYPE_DHT_FIND_PEER: /* Check if closest and not
started by us, check options, add to requests seen */
+ increment_stats (STAT_FIND_PEER);
+ if (((msg_ctx->hop_count > 0) &&
+ (0 !=
+ memcmp (&msg_ctx->peer, &my_identity,
+ sizeof (struct GNUNET_PeerIdentity)))) ||
+ (msg_ctx->client != NULL))
+ {
+ cache_response (msg_ctx);
+ if ((msg_ctx->closest == GNUNET_YES) ||
+ (msg_ctx->msg_options == GNUNET_DHT_RO_DEMULTIPLEX_EVERYWHERE))
+ handle_dht_find_peer (msg, msg_ctx);
+ }
+ else
+ route_message (msg, msg_ctx);
+ break;
+ default:
+ GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+ "`%s': Message type (%d) not handled, forwarding anyway!\n",
+ "DHT", ntohs (msg->type));
+ route_message (msg, msg_ctx);
+ }
+}
+
+
+/**
+ * Iterator over hash map entries.
+ *
+ * @param cls closure
+ * @param key current key code
+ * @param value value in the hash map
+ * @return GNUNET_YES if we should continue to
+ * iterate,
+ * GNUNET_NO if not.
+ */
+static int
+add_known_to_bloom (void *cls, const GNUNET_HashCode * key, void *value)
+{
+ struct GNUNET_CONTAINER_BloomFilter *bloom = cls;
+
+ GNUNET_CONTAINER_bloomfilter_add (bloom, key);
+ return GNUNET_YES;
+}
+
+/**
+ * Task to send a find peer message for our own peer identifier
+ * so that we can find the closest peers in the network to ourselves
+ * and attempt to connect to them.
+ *
+ * @param cls closure for this task
+ * @param tc the context under which the task is running
+ */
+static void
+send_find_peer_message (void *cls,
+ const struct GNUNET_SCHEDULER_TaskContext *tc)
+{
+ struct GNUNET_DHT_FindPeerMessage *find_peer_msg;
+ struct DHT_MessageContext msg_ctx;
+ struct GNUNET_TIME_Relative next_send_time;
+ struct GNUNET_CONTAINER_BloomFilter *temp_bloom;
+
+ if ((tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN) != 0)
+ return;
+
+ if (newly_found_peers > bucket_size) /* If we are finding peers
already, no need to send out our request right now! */
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+ "Have %d newly found peers since last find peer message
sent!\n",
+ newly_found_peers);
+ GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_MINUTES,
+ &send_find_peer_message, NULL);
+ newly_found_peers = 0;
+ return;
+ }
+
+ increment_stats (STAT_FIND_PEER_START);
+#if FIND_PEER_WITH_HELLO
+ find_peer_msg =
+ GNUNET_malloc (sizeof (struct GNUNET_DHT_FindPeerMessage) +
+ GNUNET_HELLO_size ((struct GNUNET_HELLO_Message *)
+ my_hello));
+ find_peer_msg->header.size =
+ htons (sizeof (struct GNUNET_DHT_FindPeerMessage) +
+ GNUNET_HELLO_size ((struct GNUNET_HELLO_Message *) my_hello));
+ memcpy (&find_peer_msg[1], my_hello,
+ GNUNET_HELLO_size ((struct GNUNET_HELLO_Message *) my_hello));
+#else
+ find_peer_msg = GNUNET_malloc (sizeof (struct GNUNET_DHT_FindPeerMessage));
+ find_peer_msg->header.size =
+ htons (sizeof (struct GNUNET_DHT_FindPeerMessage));
+#endif
+ find_peer_msg->header.type = htons (GNUNET_MESSAGE_TYPE_DHT_FIND_PEER);
+ temp_bloom =
+ GNUNET_CONTAINER_bloomfilter_init (NULL, DHT_BLOOM_SIZE, DHT_BLOOM_K);
+ GNUNET_CONTAINER_multihashmap_iterate (all_known_peers, &add_known_to_bloom,
+ temp_bloom);
+ GNUNET_assert (GNUNET_OK ==
+ GNUNET_CONTAINER_bloomfilter_get_raw_data (temp_bloom,
+ find_peer_msg->
+ bloomfilter,
+ DHT_BLOOM_SIZE));
+ GNUNET_CONTAINER_bloomfilter_free (temp_bloom);
+ memset (&msg_ctx, 0, sizeof (struct DHT_MessageContext));
+ memcpy (&msg_ctx.key, &my_identity.hashPubKey, sizeof (GNUNET_HashCode));
+ msg_ctx.unique_id =
+ GNUNET_ntohll (GNUNET_CRYPTO_random_u64
+ (GNUNET_CRYPTO_QUALITY_STRONG, UINT64_MAX));
+ msg_ctx.replication = DHT_DEFAULT_FIND_PEER_REPLICATION;
+ msg_ctx.msg_options = GNUNET_DHT_RO_DEMULTIPLEX_EVERYWHERE;
+ msg_ctx.network_size = log_of_network_size_estimate;
+ msg_ctx.peer = my_identity;
+ msg_ctx.importance = DHT_DEFAULT_FIND_PEER_IMPORTANCE;
+ msg_ctx.timeout = DHT_DEFAULT_FIND_PEER_TIMEOUT;
+
+ demultiplex_message (&find_peer_msg->header, &msg_ctx);
+ GNUNET_free (find_peer_msg);
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "`%s:%s': Sent `%s' request to some (?) peers\n", my_short_id,
+ "DHT", "FIND PEER");
+ if (newly_found_peers < bucket_size)
+ {
+ next_send_time.rel_value =
+ (DHT_MAXIMUM_FIND_PEER_INTERVAL.rel_value / 2) +
+ GNUNET_CRYPTO_random_u64 (GNUNET_CRYPTO_QUALITY_STRONG,
+ DHT_MAXIMUM_FIND_PEER_INTERVAL.rel_value /
2);
+ }
+ else
+ {
+ next_send_time.rel_value =
+ DHT_MINIMUM_FIND_PEER_INTERVAL.rel_value +
+ GNUNET_CRYPTO_random_u64 (GNUNET_CRYPTO_QUALITY_STRONG,
+ DHT_MAXIMUM_FIND_PEER_INTERVAL.rel_value -
+ DHT_MINIMUM_FIND_PEER_INTERVAL.rel_value);
+ }
+
+ GNUNET_assert (next_send_time.rel_value != 0);
+ find_peer_context.count = 0;
+ newly_found_peers = 0;
+ find_peer_context.start = GNUNET_TIME_absolute_get ();
+ GNUNET_SCHEDULER_add_delayed (next_send_time, &send_find_peer_message,
+ NULL);
+}
+
+
+/**
+ * Core handler for p2p route requests.
+ *
+ * @param cls closure
+ * @param message message
+ * @param peer peer identity this notification is about
+ * @param atsi performance data
+ * @return GNUNET_OK to keep the connection open,
+ * GNUNET_SYSERR to close it (signal serious error)
+ */
+static int
+handle_dht_p2p_route_request (void *cls, const struct GNUNET_PeerIdentity
*peer,
+ const struct GNUNET_MessageHeader *message,
+ const struct GNUNET_TRANSPORT_ATS_Information
+ *atsi)
+{
+#if DEBUG_DHT
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "`%s:%s': Received P2P request from peer %s\n", my_short_id,
+ "DHT", GNUNET_i2s (peer));
+#endif
+ struct GNUNET_DHT_P2PRouteMessage *incoming =
+ (struct GNUNET_DHT_P2PRouteMessage *) message;
+ struct GNUNET_MessageHeader *enc_msg =
+ (struct GNUNET_MessageHeader *) &incoming[1];
+ struct DHT_MessageContext *msg_ctx;
+ char *route_path;
+ int path_size;
+
+ if (ntohs (enc_msg->size) >= GNUNET_SERVER_MAX_MESSAGE_SIZE - 1)
+ {
+ GNUNET_break_op (0);
+ return GNUNET_YES;
+ }
+
+ if (get_max_send_delay ().rel_value > MAX_REQUEST_TIME.rel_value)
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Sending of previous replies took too long, backing off!\n");
+ increment_stats ("# route requests dropped due to high load");
+ decrease_max_send_delay (get_max_send_delay ());
+ return GNUNET_YES;
+ }
+ msg_ctx = GNUNET_malloc (sizeof (struct DHT_MessageContext));
+ msg_ctx->bloom =
+ GNUNET_CONTAINER_bloomfilter_init (incoming->bloomfilter, DHT_BLOOM_SIZE,
+ DHT_BLOOM_K);
+ GNUNET_assert (msg_ctx->bloom != NULL);
+ msg_ctx->hop_count = ntohl (incoming->hop_count);
+ memcpy (&msg_ctx->key, &incoming->key, sizeof (GNUNET_HashCode));
+ msg_ctx->replication = ntohl (incoming->desired_replication_level);
+ msg_ctx->msg_options = ntohl (incoming->options);
+ if (GNUNET_DHT_RO_RECORD_ROUTE ==
+ (msg_ctx->msg_options & GNUNET_DHT_RO_RECORD_ROUTE))
+ {
+ path_size =
+ ntohl (incoming->outgoing_path_length) *
+ sizeof (struct GNUNET_PeerIdentity);
+ if (ntohs (message->size) !=
+ (sizeof (struct GNUNET_DHT_P2PRouteMessage) + ntohs (enc_msg->size) +
+ path_size))
+ {
+ GNUNET_break_op (0);
+ GNUNET_free (msg_ctx);
+ return GNUNET_YES;
+ }
+ route_path = (char *) &incoming[1];
+ route_path = route_path + ntohs (enc_msg->size);
+ msg_ctx->path_history =
+ GNUNET_malloc (sizeof (struct GNUNET_PeerIdentity) + path_size);
+ memcpy (msg_ctx->path_history, route_path, path_size);
+ memcpy (&msg_ctx->path_history[path_size], &my_identity,
+ sizeof (struct GNUNET_PeerIdentity));
+ msg_ctx->path_history_len = ntohl (incoming->outgoing_path_length) + 1;
+ }
+ msg_ctx->network_size = ntohl (incoming->network_size);
+ msg_ctx->peer = *peer;
+ msg_ctx->importance = DHT_DEFAULT_P2P_IMPORTANCE;
+ msg_ctx->timeout = DHT_DEFAULT_P2P_TIMEOUT;
+ demultiplex_message (enc_msg, msg_ctx);
+ if (msg_ctx->bloom != NULL)
+ {
+ GNUNET_CONTAINER_bloomfilter_free (msg_ctx->bloom);
+ msg_ctx->bloom = NULL;
+ }
+ GNUNET_free (msg_ctx);
+ return GNUNET_YES;
+}
+
+
+/**
+ * Core handler for p2p route results.
+ *
+ * @param cls closure
+ * @param message message
+ * @param peer peer identity this notification is about
+ * @param atsi performance data
+ *
+ */
+static int
+handle_dht_p2p_route_result (void *cls, const struct GNUNET_PeerIdentity *peer,
+ const struct GNUNET_MessageHeader *message,
+ const struct GNUNET_TRANSPORT_ATS_Information
+ *atsi)
+{
+#if DEBUG_DHT
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "`%s:%s': Received request from peer %s\n", my_short_id, "DHT",
+ GNUNET_i2s (peer));
+#endif
+ const struct GNUNET_DHT_P2PRouteResultMessage *incoming =
+ (const struct GNUNET_DHT_P2PRouteResultMessage *) message;
+ struct GNUNET_MessageHeader *enc_msg =
+ (struct GNUNET_MessageHeader *) &incoming[1];
+ struct DHT_MessageContext msg_ctx;
+
+ if (ntohs (enc_msg->size) >= GNUNET_SERVER_MAX_MESSAGE_SIZE - 1)
+ {
+ GNUNET_break_op (0);
+ return GNUNET_YES;
+ }
+
+ memset (&msg_ctx, 0, sizeof (struct DHT_MessageContext));
+ memcpy (&msg_ctx.key, &incoming->key, sizeof (GNUNET_HashCode));
+ msg_ctx.msg_options = ntohl (incoming->options);
+ msg_ctx.hop_count = ntohl (incoming->hop_count);
+ msg_ctx.peer = *peer;
+ msg_ctx.importance = DHT_DEFAULT_P2P_IMPORTANCE + 2; /* Make result routing
a higher priority */
+ msg_ctx.timeout = DHT_DEFAULT_P2P_TIMEOUT;
+ if ((GNUNET_DHT_RO_RECORD_ROUTE ==
+ (msg_ctx.msg_options & GNUNET_DHT_RO_RECORD_ROUTE)) &&
+ (ntohl (incoming->outgoing_path_length) > 0))
+ {
+ if (ntohs (message->size) -
+ sizeof (struct GNUNET_DHT_P2PRouteResultMessage) -
+ ntohs (enc_msg->size) !=
+ ntohl (incoming->outgoing_path_length) *
+ sizeof (struct GNUNET_PeerIdentity))
+ {
+#if DEBUG_DHT
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Return message indicated a path was included, but sizes are
wrong: Total size %d, enc size %d, left %d, expected %d\n",
+ ntohs (message->size), ntohs (enc_msg->size),
+ ntohs (message->size) -
+ sizeof (struct GNUNET_DHT_P2PRouteResultMessage) -
+ ntohs (enc_msg->size),
+ ntohl (incoming->outgoing_path_length) *
+ sizeof (struct GNUNET_PeerIdentity));
+#endif
+ GNUNET_break_op (0);
+ return GNUNET_NO;
+ }
+ msg_ctx.path_history = (char *) &incoming[1];
+ msg_ctx.path_history += ntohs (enc_msg->size);
+ msg_ctx.path_history_len = ntohl (incoming->outgoing_path_length);
+ }
+ route_result_message (enc_msg, &msg_ctx);
+ return GNUNET_YES;
+}
+
+
+/**
+ * Receive the HELLO from transport service,
+ * free current and replace if necessary.
+ *
+ * @param cls NULL
+ * @param message HELLO message of peer
+ */
+static void
+process_hello (void *cls, const struct GNUNET_MessageHeader *message)
+{
+#if DEBUG_DHT
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Received our `%s' from transport service\n", "HELLO");
+#endif
+
+ GNUNET_assert (message != NULL);
+ GNUNET_free_non_null (my_hello);
+ my_hello = GNUNET_malloc (ntohs (message->size));
+ memcpy (my_hello, message, ntohs (message->size));
+}
+
+
+/**
+ * Task run during shutdown.
+ *
+ * @param cls unused
+ * @param tc unused
+ */
+static void
+shutdown_task (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
+{
+ int bucket_count;
+ struct PeerInfo *pos;
+
+ if (NULL != ghh)
+ {
+ GNUNET_TRANSPORT_get_hello_cancel (ghh);
+ ghh = NULL;
+ }
+ if (transport_handle != NULL)
+ {
+ GNUNET_free_non_null (my_hello);
+ GNUNET_TRANSPORT_disconnect (transport_handle);
+ transport_handle = NULL;
+ }
+ GDS_NSE_done ();
+ if (coreAPI != NULL)
+ {
+#if DEBUG_DHT
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "%s:%s Disconnecting core!\n",
+ my_short_id, "DHT");
+#endif
+ GNUNET_CORE_disconnect (coreAPI);
+ coreAPI = NULL;
+ }
+ for (bucket_count = lowest_bucket; bucket_count < MAX_BUCKETS;
bucket_count++)
+ {
+ while (k_buckets[bucket_count].head != NULL)
+ {
+ pos = k_buckets[bucket_count].head;
+#if DEBUG_DHT
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "%s:%s Removing peer %s from bucket %d!\n", my_short_id,
+ "DHT", GNUNET_i2s (&pos->id), bucket_count);
+#endif
+ delete_peer (pos, bucket_count);
+ }
+ }
+ if (datacache != NULL)
+ {
+#if DEBUG_DHT
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "%s:%s Destroying datacache!\n",
+ my_short_id, "DHT");
+#endif
+ GNUNET_DATACACHE_destroy (datacache);
+ datacache = NULL;
+ }
+ if (stats != NULL)
+ {
+ GNUNET_STATISTICS_destroy (stats, GNUNET_YES);
+ stats = NULL;
+ }
+ if (block_context != NULL)
+ {
+ GNUNET_BLOCK_context_destroy (block_context);
+ block_context = NULL;
+ }
+ GNUNET_free_non_null (my_short_id);
+ my_short_id = NULL;
+}
+
+
+/**
+ * To be called on core init/fail.
+ *
+ * @param cls service closure
+ * @param server handle to the server for this service
+ * @param identity the public identity of this peer
+ * @param publicKey the public key of this peer
+ */
+static void
+core_init (void *cls, struct GNUNET_CORE_Handle *server,
+ const struct GNUNET_PeerIdentity *identity,
+ const struct GNUNET_CRYPTO_RsaPublicKeyBinaryEncoded *publicKey)
+{
+
+ if (server == NULL)
+ {
+#if DEBUG_DHT
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "%s: Connection to core FAILED!\n",
+ "dht", GNUNET_i2s (identity));
+#endif
+ GNUNET_SCHEDULER_cancel (cleanup_task);
+ GNUNET_SCHEDULER_add_now (&shutdown_task, NULL);
+ return;
+ }
+#if DEBUG_DHT
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "%s: Core connection initialized, I am peer: %s\n", "dht",
+ GNUNET_i2s (identity));
+#endif
+
+ /* Copy our identity so we can use it */
+ memcpy (&my_identity, identity, sizeof (struct GNUNET_PeerIdentity));
+ if (my_short_id != NULL)
+ GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+ "%s Receive CORE INIT message but have already been
initialized! Did CORE fail?\n",
+ "DHT SERVICE");
+ my_short_id = GNUNET_strdup (GNUNET_i2s (&my_identity));
+}
+
+
+static struct GNUNET_CORE_MessageHandler core_handlers[] = {
+ {&handle_dht_p2p_route_request, GNUNET_MESSAGE_TYPE_DHT_P2P_ROUTE, 0},
+ {&handle_dht_p2p_route_result, GNUNET_MESSAGE_TYPE_DHT_P2P_ROUTE_RESULT, 0},
+ {NULL, 0, 0}
+};
+
+
+
+
+/**
+ * Process dht requests.
+ *
+ * @param cls closure
+ * @param server the initialized server
+ * @param c configuration to use
+ */
+static void
+run (void *cls, struct GNUNET_SERVER_Handle *server,
+ const struct GNUNET_CONFIGURATION_Handle *c)
+{
+ struct GNUNET_TIME_Relative next_send_time;
+ unsigned long long temp_config_num;
+
+ cfg = c;
+ datacache = GNUNET_DATACACHE_create (cfg, "dhtcache");
+ coreAPI = GNUNET_CORE_connect (cfg, /* Main configuration */
+ DEFAULT_CORE_QUEUE_SIZE, /* queue size
*/
+ NULL, /* Closure passed to DHT functions */
+ &core_init, /* Call core_init once
connected */
+ &handle_core_connect, /* Handle connects */
+ &handle_core_disconnect, /* remove
peers on disconnects */
+ NULL, /* Do we care about "status" updates?
*/
+ NULL, /* Don't want notified about all
incoming messages */
+ GNUNET_NO, /* For header only inbound
notification */
+ NULL, /* Don't want notified about all
outbound messages */
+ GNUNET_NO, /* For header only outbound
notification */
+ core_handlers); /* Register these
handlers */
+
+ if (coreAPI == NULL)
+ return;
+ transport_handle =
+ GNUNET_TRANSPORT_connect (cfg, NULL, NULL, NULL, NULL, NULL);
+ if (transport_handle != NULL)
+ ghh = GNUNET_TRANSPORT_get_hello (transport_handle, &process_hello, NULL);
+ else
+ GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+ "Failed to connect to transport service!\n");
+ block_context = GNUNET_BLOCK_context_create (cfg);
+ lowest_bucket = MAX_BUCKETS - 1;
+ all_known_peers = GNUNET_CONTAINER_multihashmap_create (MAX_BUCKETS / 8);
+ GNUNET_assert (all_known_peers != NULL);
+
+ if (GNUNET_OK ==
+ GNUNET_CONFIGURATION_get_value_number (cfg, "DHT", "bucket_size",
+ &temp_config_num))
+ {
+ bucket_size = (unsigned int) temp_config_num;
+ }
+
+ stats = GNUNET_STATISTICS_create ("dht", cfg);
+ next_send_time.rel_value =
+ DHT_MINIMUM_FIND_PEER_INTERVAL.rel_value +
+ GNUNET_CRYPTO_random_u64 (GNUNET_CRYPTO_QUALITY_STRONG,
+ (DHT_MAXIMUM_FIND_PEER_INTERVAL.rel_value /
+ 2) -
+ DHT_MINIMUM_FIND_PEER_INTERVAL.rel_value);
+ find_peer_context.start = GNUNET_TIME_absolute_get ();
+ GNUNET_SCHEDULER_add_delayed (next_send_time, &send_find_peer_message,
+ &find_peer_context);
+
+ /* Scheduled the task to clean up when shutdown is called */
+ cleanup_task =
+ GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_FOREVER_REL,
+ &shutdown_task, NULL);
+}
+
+
+/**
+ * The main function for the dht service.
+ *
+ * @param argc number of arguments from the command line
+ * @param argv command line arguments
+ * @return 0 ok, 1 on error
+ */
+int
+main (int argc, char *const *argv)
+{
+ int ret;
+ struct RecentRequest *recent_req;
+
+ recent_heap =
+ GNUNET_CONTAINER_heap_create (GNUNET_CONTAINER_HEAP_ORDER_MIN);
+ recent_find_peer_requests =
+ GNUNET_CONTAINER_multihashmap_create (MAX_BUCKETS / 8);
+ ret =
+ (GNUNET_OK ==
+ GNUNET_SERVICE_run (argc, argv, "dht", GNUNET_SERVICE_OPTION_NONE, &run,
+ NULL)) ? 0 : 1;
+ while (GNUNET_CONTAINER_heap_get_size (recent_heap) > 0)
+ {
+ recent_req = GNUNET_CONTAINER_heap_peek (recent_heap);
+ GNUNET_assert (recent_req != NULL);
+ GNUNET_CONTAINER_heap_remove_node (recent_req->heap_node);
+ GNUNET_CONTAINER_bloomfilter_free (recent_req->bloom);
+ GNUNET_free (recent_req);
+ }
+ GNUNET_CONTAINER_heap_destroy (recent_heap);
+ recent_heap = NULL;
+ GNUNET_CONTAINER_multihashmap_destroy (recent_find_peer_requests);
+ recent_find_peer_requests = NULL;
+ return ret;
+}
+
+/* end of gnunet-service-dht.c */
Added: gnunet/src/dht/gnunet-service-dht_clients.c
===================================================================
--- gnunet/src/dht/gnunet-service-dht_clients.c (rev 0)
+++ gnunet/src/dht/gnunet-service-dht_clients.c 2011-09-21 05:51:38 UTC (rev
16976)
@@ -0,0 +1,876 @@
+/*
+ This file is part of GNUnet.
+ (C) 2009, 2010, 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
+ 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
+ WITHOUT ANY WARRANTY; without even the implied warranty of
+ MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ General Public License for more details.
+
+ 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.
+*/
+
+/**
+ * @file dht/gnunet-service-dht_clients.c
+ * @brief GNUnet DHT service's client management code
+ * @author Christian Grothoff
+ * @author Nathan Evans
+ */
+
+#include "platform.h"
+#include "gnunet_block_lib.h"
+#include "gnunet_util_lib.h"
+#include "gnunet_protocols.h"
+#include "gnunet_nse_service.h"
+#include "gnunet_core_service.h"
+#include "gnunet_datacache_lib.h"
+#include "gnunet_transport_service.h"
+#include "gnunet_hello_lib.h"
+#include "gnunet_dht_service.h"
+#include "gnunet_statistics_service.h"
+#include "dht_new.h"
+#include <fenv.h>
+#include "gnunet-service-dht_clients.h"
+#include "gnunet-service-dht_neighbours.h"
+
+
+/**
+ * Linked list of messages to send to clients.
+ */
+struct PendingMessage
+{
+ /**
+ * Pointer to next item in the list
+ */
+ struct PendingMessage *next;
+
+ /**
+ * Pointer to previous item in the list
+ */
+ struct PendingMessage *prev;
+
+ /**
+ * Actual message to be sent, allocated at the end of the struct:
+ * // msg = (cast) &pm[1];
+ * // memcpy (&pm[1], data, len);
+ */
+ const struct GNUNET_MessageHeader *msg;
+
+};
+
+
+/**
+ * Struct containing information about a client,
+ * handle to connect to it, and any pending messages
+ * that need to be sent to it.
+ */
+struct ClientList
+{
+ /**
+ * Linked list of active clients
+ */
+ struct ClientList *next;
+
+ /**
+ * Linked list of active clients
+ */
+ struct ClientList *prev;
+
+ /**
+ * The handle to this client
+ */
+ struct GNUNET_SERVER_Client *client_handle;
+
+ /**
+ * Handle to the current transmission request, NULL
+ * if none pending.
+ */
+ struct GNUNET_CONNECTION_TransmitHandle *transmit_handle;
+
+ /**
+ * Linked list of pending messages for this client
+ */
+ struct PendingMessage *pending_head;
+
+ /**
+ * Tail of linked list of pending messages for this client
+ */
+ struct PendingMessage *pending_tail;
+
+};
+
+
+/**
+ * Entry in the DHT routing table for a client's GET request.
+ */
+struct ClientQueryRecord
+{
+
+ /**
+ * The key this request was about
+ */
+ GNUNET_HashCode key;
+
+ /**
+ * Client responsible for the request.
+ */
+ struct ClientList *client;
+
+ /**
+ * Extended query (see gnunet_block_lib.h), allocated at the end of this
struct.
+ */
+ const void *xquery;
+
+ /**
+ * Replies we have already seen for this request.
+ */
+ GNUNET_HashCode *seen_replies;
+
+ /**
+ * Pointer to this nodes heap location in the retry-heap (for fast removal)
+ */
+ struct GNUNET_CONTAINER_HeapNode *hnode;
+
+ /**
+ * What's the delay between re-try operations that we currently use for this
+ * request?
+ */
+ struct GNUNET_TIME_Relative retry_frequency;
+
+ /**
+ * What's the next time we should re-try this request?
+ */
+ struct GNUNET_TIME_Absolute retry_time;
+
+ /**
+ * The unique identifier of this request
+ */
+ uint64_t unique_id;
+
+ /**
+ * Number of bytes in xquery.
+ */
+ size_t xquery_size;
+
+ /**
+ * Number of entries in 'seen_replies'.
+ */
+ unsigned int seen_replies_count;
+
+ /**
+ * Desired replication level
+ */
+ uint32_t replication;
+
+ /**
+ * Any message options for this request
+ */
+ uint32_t msg_options;
+
+ /**
+ * The type for the data for the GET request; actually an 'enum
+ * GNUNET_BLOCK_Type'.
+ */
+ uint32_t msg_type;
+
+};
+
+
+/**
+ * List of active clients.
+ */
+static struct ClientList *client_head;
+
+/**
+ * List of active clients.
+ */
+static struct ClientList *client_tail;
+
+/**
+ * Hashmap for fast key based lookup, maps keys to 'struct ClientQueryRecord'
entries.
+ */
+static struct GNUNET_CONTAINER_MultiHashMap *forward_map;
+
+/**
+ * Heap with all of our client's request, sorted by retry time (earliest on
top).
+ */
+static struct GNUNET_CONTAINER_Heap *retry_heap;
+
+/**
+ * Task that re-transmits requests (using retry_heap).
+ */
+static GNUNET_SCHEDULER_TaskIdentifier retry_task;
+
+
+/**
+ * Find a client if it exists, add it otherwise.
+ *
+ * @param client the server handle to the client
+ *
+ * @return the client if found, a new client otherwise
+ */
+static struct ClientList *
+find_active_client (struct GNUNET_SERVER_Client *client)
+{
+ struct ClientList *pos = client_list;
+ struct ClientList *ret;
+
+ while (pos != NULL)
+ {
+ if (pos->client_handle == client)
+ return pos;
+ pos = pos->next;
+ }
+ ret = GNUNET_malloc (sizeof (struct ClientList));
+ ret->client_handle = client;
+ GNUNET_CONTAINER_DLL_insert (client_head,
+ client_tail,
+ ret);
+ return ret;
+}
+
+
+/**
+ * Iterator over hash map entries that frees all entries
+ * associated with the given client.
+ *
+ * @param cls client to search for in source routes
+ * @param key current key code (ignored)
+ * @param value value in the hash map, a ClientQueryRecord
+ * @return GNUNET_YES (we should continue to iterate)
+ */
+static int
+remove_client_records (void *cls, const GNUNET_HashCode * key, void *value)
+{
+ struct ClientList *client = cls;
+ struct ClientQueryRecord *record = value;
+
+ if (record->client != client)
+ return GNUNET_YES;
+ GNUNET_assert (GNUNET_YES ==
+ GNUNET_CONTAINER_multihashmap_remove (forward_map,
+ key, record));
+ GNUNET_CONTAINER_heap_remove_node (record->hnode);
+ GNUNET_ARRAY_append (record->seen_replies,
+ record->seen_replies_count,
+ 0);
+ GNUNET_free (record);
+ return GNUNET_YES;
+}
+
+
+/**
+ * Functions with this signature are called whenever a client
+ * is disconnected on the network level.
+ *
+ * @param cls closure (NULL for dht)
+ * @param client identification of the client; NULL
+ * for the last call when the server is destroyed
+ */
+static void
+handle_client_disconnect (void *cls,
+ struct GNUNET_SERVER_Client *client)
+{
+ struct ClientList *pos = client_list;
+ struct ClientList *found;
+ struct PendingMessage *reply;
+
+ found = NULL;
+ while (pos != NULL)
+ {
+ if (pos->client_handle == client)
+ {
+ GNUNET_CONTAINER_DLL_remove (client_head,
+ client_tail,
+ pos);
+ found = pos;
+ break;
+ }
+ pos = pos->next;
+ }
+ if (found == NULL)
+ return;
+ if (found->transmit_handle != NULL)
+ GNUNET_CONNECTION_notify_transmit_ready_cancel (found->transmit_handle);
+ while (NULL != (reply = found->pending_head))
+ {
+ GNUNET_CONTAINER_DLL_remove (found->pending_head, found->pending_tail,
+ reply);
+ GNUNET_free (reply);
+ }
+ GNUNET_CONTAINER_multihashmap_iterate (forward_list.hashmap,
+ &remove_client_records, found);
+ GNUNET_free (found);
+}
+
+
+/**
+ * Route the given request via the DHT. This includes updating
+ * the bloom filter and retransmission times, building the P2P
+ * message and initiating the routing operation.
+ */
+static void
+transmit_request (struct ClientQueryRecord *cqr)
+{
+ int32_t reply_bf_mutator;
+
+ reply_bf_mutator = (int32_t) GNUNET_CRYPTO_random_u32
(GNUNET_CRYPTO_QUALITY_WEAK,
+ UINT32_MAX);
+ reply_bf = GNUNET_BLOCK_construct_bloomfilter (reply_bf_mutator,
+ cqr->seen_replies,
+ cqr->seen_replies_count);
+ GST_NEIGHBOURS_handle_get (cqr->msg_type,
+ cqr->msg_options,
+ cqr->replication,
+ &cqr->key,
+ cqr->xquery,
+ cqr->xquery_size,
+ reply_bf,
+ reply_bf_mutator,
+ NULL /* no peers blocked initially */);
+ GNUNET_CONTAINER_bloomfilter_destroy (reply_bf);
+
+ /* exponential back-off for retries, max 1h */
+ cqr->retry_frequency =
+ GNUNET_TIME_relative_max (GNUNET_TIME_UNIT_HOURS,
+ GNUNET_TIME_relative_multiply
(cqr->retry_frequency, 2));
+ cqr->retry_time = GNUNET_TIME_relative_to_absolute (cqr->retry_frequency);
+}
+
+
+/**
+ * Task that looks at the 'retry_heap' and transmits all of the requests
+ * on the heap that are ready for transmission. Then re-schedules
+ * itself (unless the heap is empty).
+ *
+ * @param cls unused
+ * @param tc scheduler context
+ */
+static void
+transmit_next_request_task (void *cls,
+ const struct GNUNET_SCHEDULER_TaskContext *tc)
+{
+ struct ClientQueryRecord *cqr;
+ struct GNUNET_TIME_Relative delay;
+
+ retry_task = GNUNET_SCHEDULER_NO_TASK;
+ if (0 != (tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN))
+ return;
+ while (NULL != (cqr = GNUNET_CONTAINER_heap_remove_root (retry_heap)))
+ {
+ cqr->hnode = NULL;
+ delay = GNUNET_TIME_absolute_get_remaining (cqr->retry_time);
+ if (delay.value > 0)
+ {
+ cqr->hnode = GNUNET_CONTAINER_heap_insert (retry_heap, cqr,
+ cqr->retry_time.abs_value);
+ retry_task = GNUNET_SCHEDULER_add_delayed (delay,
+
&transmit_next_request_task,
+ NULL);
+ return;
+ }
+ transmit_request (cqr);
+ }
+}
+
+
+/**
+ * Handler for PUT messages.
+ *
+ * @param cls closure for the service
+ * @param client the client we received this message from
+ * @param message the actual message received
+ */
+static void
+handle_dht_local_put (void *cls, struct GNUNET_SERVER_Client *client,
+ const struct GNUNET_MessageHeader *message)
+{
+ const struct GNUNET_DHT_ClientPutMessage *dht_msg;
+ uint16_t size;
+
+ size = ntohs (message->size);
+ if (size < sizeof (struct GNUNET_DHT_ClientPutMessage))
+ {
+ GNUNET_break (0);
+ GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
+ }
+ dht_msg = (const struct GNUNET_DHT_ClientPutMessage *) message;
+ GST_NEIGHBOURS_handle_put (ntohl (dht_msg->type),
+ ntohl (dht_msg->options),
+ ntohl (dht_msg->desired_replication_level),
+ GNUNET_TIME_absolute_ntoh (dht_msg->expiration),
+ &dht_msg->key,
+ &dht_msg[1],
+ size - sizeof (struct
GNUNET_DHT_ClientPutMessage));
+ GNUNET_SERVER_receive_done (client, GNUNET_OK);
+}
+
+
+/**
+ * Handler for any generic DHT messages, calls the appropriate handler
+ * depending on message type, sends confirmation if responses aren't otherwise
+ * expected.
+ *
+ * @param cls closure for the service
+ * @param client the client we received this message from
+ * @param message the actual message received
+ */
+static void
+handle_dht_local_get (void *cls, struct GNUNET_SERVER_Client *client,
+ const struct GNUNET_MessageHeader *message)
+{
+ const struct GNUNET_DHT_ClientGetMessage *get;
+ const struct GNUNET_MessageHeader *enc_msg;
+
+ struct ClientQueryRecord *cqr;
+ size_t xquery_size;
+ const char* xquery;
+ uint16_t size;
+
+ size = ntohs (message->size);
+ if (size < sizeof (struct GNUNET_DHT_ClientGetMessage))
+ {
+ GNUNET_break (0);
+ GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
+ return;
+ }
+ xquery_size = size - sizeof (struct GNUNET_DHT_ClientGetMessage);
+ get = (const struct GNUNET_DHT_ClientGetMessage *) message;
+ xquery = (const char*) &get[1];
+
+ cqr = GNUNET_malloc (sizeof (struct ClientQueryRecord) + xquery_size);
+ cqr->key = get->key;
+ cqr->client = find_active_client (client);
+ cqr->xquery = (void*) &cqr[1];
+ memcpy (&cqr[1], xquery, xquery_size);
+ cqr->hnode = GNUNET_CONTAINER_heap_insert (retry_heap, cqr, 0);
+ cqr->retry_frequency = GNUNET_TIME_UNIT_MILLISECONDS;
+ cqr->retry_time = GNUNET_TIME_absolute_get ();
+ cqr->unique_id = get->unique_id;
+ cqr->xquery_size = xquery_size;
+ cqr->replication = ntohl (get->desired_replication_level);
+ cqr->msg_options = ntohl (get->options);
+ cqr->msg_type = ntohl (get->type);
+ GNUNET_CONTAINER_multihashmap_put (forward_map, KEY, cqr,
+
GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
+ if (GNUNET_SCHEDULER_NO_TASK != retry_task)
+ GNUNET_SCHEDULER_cancel (retry_task);
+ retry_task = GNUNET_SCHEDULER_add_now (&transmit_next_request_task, NULL);
+ GNUNET_SERVER_receive_done (client, GNUNET_OK);
+}
+
+
+/**
+ * Closure for 'remove_by_uid'.
+ */
+struct RemoveByUidContext
+{
+ /**
+ * Client that issued the removal request.
+ */
+ struct ClientList *client;
+
+ /**
+ * Unique ID of the request.
+ */
+ uint64_t uid;
+};
+
+
+/**
+ * Iterator over hash map entries that frees all entries
+ * that match the given client and UID.
+ *
+ * @param cls UID and client to search for in source routes
+ * @param key current key code
+ * @param value value in the hash map, a ClientQueryRecord
+ * @return GNUNET_YES (we should continue to iterate)
+ */
+static int
+remove_by_uid (void *cls, const GNUNET_HashCode * key, void *value)
+{
+ const struct RemoveByUidContext *ctx = cls;
+ struct ClientQueryRecord *record = value;
+
+ if (record->uid != ctx->uid)
+ return GNUNET_YES;
+ return remove_client_records (ctx->client, key, record);
+}
+
+
+/**
+ * Handler for any generic DHT stop messages, calls the appropriate handler
+ * depending on message type (if processed locally)
+ *
+ * @param cls closure for the service
+ * @param client the client we received this message from
+ * @param message the actual message received
+ *
+ */
+static void
+handle_dht_local_get_stop (void *cls, struct GNUNET_SERVER_Client *client,
+ const struct GNUNET_MessageHeader *message)
+{
+ const struct GNUNET_DHT_ClientGetStopMessage *dht_stop_msg =
+ (const struct GNUNET_DHT_ClientGetStopMessage *) message;
+
+ ctx.client = find_active_client (client);
+ ctx.uid = &dht_stop_msg.unique_id);
+ GNUNET_CONTAINER_multihashmap_get_multiple (forward_map,
+ &dht_stop_msg->key,
+ &remove_by_uid,
+ &ctx);
+ GNUNET_SERVER_receive_done (client, GNUNET_OK);
+}
+
+
+/**
+ * Task run to check for messages that need to be sent to a client.
+ *
+ * @param client a ClientList, containing the client and any messages to be
sent to it
+ */
+static void
+process_pending_messages (struct ClientList *client);
+
+
+/**
+ * Callback called as a result of issuing a GNUNET_SERVER_notify_transmit_ready
+ * request. A ClientList is passed as closure, take the head of the list
+ * and copy it into buf, which has the result of sending the message to the
+ * client.
+ *
+ * @param cls closure to this call
+ * @param size maximum number of bytes available to send
+ * @param buf where to copy the actual message to
+ *
+ * @return the number of bytes actually copied, 0 indicates failure
+ */
+static size_t
+send_reply_to_client (void *cls, size_t size, void *buf)
+{
+ struct ClientList *client = cls;
+ char *cbuf = buf;
+ struct PendingMessage *reply;
+ size_t off;
+ size_t msize;
+
+ client->transmit_handle = NULL;
+ if (buf == NULL)
+ {
+ /* client disconnected */
+ return 0;
+ }
+ off = 0;
+ while ((NULL != (reply = client->pending_head)) &&
+ (size >= off + (msize = ntohs (reply->msg->size))))
+ {
+ GNUNET_CONTAINER_DLL_remove (client->pending_head, client->pending_tail,
+ reply);
+ memcpy (&cbuf[off], reply->msg, msize);
+ GNUNET_free (reply);
+ off += msize;
+ }
+ process_pending_messages (client);
+ return off;
+}
+
+
+/**
+ * Task run to check for messages that need to be sent to a client.
+ *
+ * @param client a ClientList, containing the client and any messages to be
sent to it
+ */
+static void
+process_pending_messages (struct ClientList *client)
+{
+ if ((client->pending_head == NULL) || (client->transmit_handle != NULL))
+ return;
+ client->transmit_handle =
+ GNUNET_SERVER_notify_transmit_ready (client->client_handle,
+ ntohs (client->pending_head->
+ msg->size),
+ GNUNET_TIME_UNIT_FOREVER_REL,
+ &send_reply_to_client, client);
+}
+
+
+/**
+ * Add a PendingMessage to the clients list of messages to be sent
+ *
+ * @param client the active client to send the message to
+ * @param pending_message the actual message to send
+ */
+static void
+add_pending_message (struct ClientList *client,
+ struct PendingMessage *pending_message)
+{
+ GNUNET_CONTAINER_DLL_insert_tail (client->pending_head, client->pending_tail,
+ pending_message);
+ process_pending_messages (client);
+}
+
+
+/**
+ * Closure for 'forward_reply'
+ */
+struct ForwardReplyContext
+{
+
+ /**
+ * Actual message to send to matching clients.
+ */
+ struct PendingMessage *pm;
+
+ /**
+ * Embedded payload.
+ */
+ const void *data;
+
+ /**
+ * Type of the data.
+ */
+ uint32_t type;
+
+ /**
+ * Number of bytes in data.
+ */
+ size_t data_size;
+
+ /**
+ * Do we need to copy 'pm' because it was already used?
+ */
+ int do_copy;
+
+};
+
+
+/**
+ * Iterator over hash map entries that send a given reply to
+ * each of the matching clients. With some tricky recycling
+ * of the buffer.
+ *
+ * @param cls the 'struct ForwardReplyContext'
+ * @param key current key
+ * @param value value in the hash map, a ClientQueryRecord
+ * @return GNUNET_YES (we should continue to iterate),
+ * if the result is mal-formed, GNUNET_NO
+ */
+static int
+forward_reply (void *cls, const GNUNET_HashCode * key, void *value)
+{
+ struct ForwardReplyContext *frc = cls;
+ struct ClientQueryRecord *record = value;
+ struct PendingMessage *pm;
+ struct ReplyMessage *reply;
+ enum GNUNET_BLOCK_EvaluationResult eval;
+ int do_free;
+ GNUNET_HashCode ch;
+ unsigned int i;
+
+ if ( (record->type != GNUNET_BLOCK_TYPE_ANY) &&
+ (record->type != frc->type) )
+ return GNUNET_YES; /* type mismatch */
+ GNUNET_CRYPTO_hash (frc->data,
+ frc->data_size,
+ &ch);
+ for (i=0;i<record->seen_replies_count;i++)
+ if (0 == memcmp (&record->seen_replies[i],
+ &ch,
+ sizeof (GNUNET_HashCode)))
+ return GNUNET_YES; /* duplicate */
+ eval =
+ GNUNET_BLOCK_evaluate (GDS_block_context,
+ record->type, key,
+ NULL, 0,
+ record->xquery,
+ record->xquery_size,
+ frc->data,
+ frc->data_size);
+ switch (eval)
+ {
+ case GNUNET_BLOCK_EVALUATION_OK_LAST:
+ do_free = GNUNET_YES;
+ break;
+ case GNUNET_BLOCK_EVALUATION_OK_MORE:
+ GNUNET_ARRAY_append (record->seen_replies,
+ record->seen_replies_count,
+ ch);
+ do_free = GNUNET_NO;
+ break;
+ case GNUNET_BLOCK_EVALUATION_OK_DUPLICATE:
+ /* should be impossible to encounter here */
+ GNUNET_break (0);
+ return GNUNET_YES;
+ case GNUNET_BLOCK_EVALUATION_RESULT_INVALID:
+ GNUNET_break_op (0);
+ return GNUNET_NO;
+ case GNUNET_BLOCK_EVALUATION_REQUEST_VALID:
+ GNUNET_break (0);
+ return GNUNET_NO;
+ case GNUNET_BLOCK_EVALUATION_REQUEST_INVALID:
+ GNUNET_break (0);
+ return GNUNET_NO;
+ case GNUNET_BLOCK_EVALUATION_TYPE_NOT_SUPPORTED:
+ GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+ "Unsupported block type (%u) in request!\n",
+ record->type);
+ return GNUNET_NO;
+ }
+ if (GNUNET_NO == frc->do_copy)
+ {
+ /* first time, we can use the original data */
+ pm = frc->pm;
+ frc->do_copy = GNUNET_YES;
+ }
+ else
+ {
+ /* two clients waiting for same reply, must copy for queueing */
+ pm = GNUNET_malloc (sizeof (struct PendingMessage) +
+ ntohs (frc->pm->msg->size));
+ memcpy (pm, frc->pm,
+ sizeof (struct PendingMessage) + ntohs (frc->pm->msg->size));
+ pm->next = pm->prev = NULL;
+ }
+ reply = (struct ReplyMessage*) &pm[1];
+ reply->unique_id = record->unique_id;
+ add_pending_message (record->client, pm);
+ if (GNUNET_YES == do_free)
+ remove_client_records (record->client, key, record);
+ return GNUNET_YES;
+}
+
+
+/**
+ * Handle a reply we've received from another peer. If the reply
+ * matches any of our pending queries, forward it to the respective
+ * client(s).
+ *
+ * @param expiration when will the reply expire
+ * @param key the query this reply is for
+ * @param get_path_length number of peers in 'get_path'
+ * @param get_path path the reply took on get
+ * @param put_path_length number of peers in 'put_path'
+ * @param put_path path the reply took on put
+ * @param type type of the reply
+ * @param data_size number of bytes in 'data'
+ * @param data application payload data
+ */
+void
+GDS_CLIENT_handle_reply (struct GNUNET_TIME_Absolute expiration,
+ const GNUNET_HashCode *key,
+ unsigned int get_path_length,
+ const struct GNUNET_PeerIdentity *get_path,
+ unsigned int put_path_length,
+ const struct GNUNET_PeerIdentity *put_path,
+ uint32_t type,
+ size_t data_size,
+ const void *data)
+{
+ struct ForwardReplyContext frc;
+ struct PendingMessage *pm;
+ struct ReplyMessage *reply;
+ struct GNUNET_PeerIdentity *paths;
+ size_t msize;
+
+ if (NULL ==
+ GNUNET_CONTAINER_multihashmap_get (foward_map, key))
+ return; /* no matching request, fast exit! */
+ msize = sizeof(struct ReplyMessage) + data_size +
+ (get_path_length + put_path_length) * sizeof (struct GNUNET_PeerIdentity);
+ if (msize >= GNUNET_SERVER_MAX_MESSAGE_SIZE)
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+ _("Could not pass reply to client, message too big!\n"));
+ return;
+ }
+ pm = (struct PendingMessage *) GNUNET_malloc (msize + sizeof (struct
PendingMessage));
+ reply = (struct ReplyMessage*) &pm[1];
+ pm->msg = &reply->header;
+ reply->header.size = htons ((uint16_t) msize);
+ reply->header.type = htons (GNUNET_MESSAGE_TYPE_DHT_CLIENT_RESULT);
+ reply->type = htonl (type);
+ reply->get_path_length = htonl (get_path_length);
+ reply->put_path_length = htonl (put_path_length);
+ reply->unique_id = 0; /* filled in later */
+ reply->expiration = GNUNET_TIME_absolute_hton (expiration);
+ reply->key = *key;
+ paths = (struct GNUNET_PeerIdentity*) &reply[1];
+ mempcy (paths, get_path,
+ sizeof (struct GNUNET_PeerIdentity) * get_path_length);
+ mempcy (&paths[get_path_length],
+ put_path, sizeof (struct GNUNET_PeerIdentity) * put_path_length);
+ memcpy (&paths[get_path_length + put_path_length],
+ data,
+ data_size);
+ frc.do_copy = GNUNET_NO;
+ frc.pm = pm;
+ frc.data = data;
+ frc.data_size = data_size;
+ frc.type = type;
+ GNUNET_CONTAINER_multihashmap_get_multiple (foward_map, key,
+ &forward_reply,
+ &frc);
+ if (GNUNET_NO == frc.do_copy)
+ {
+ /* did not match any of the requests, free! */
+ GNUNET_free (buf);
+ }
+}
+
+
+/**
+ * Initialize client subsystem.
+ *
+ * @param server the initialized server
+ */
+void
+GDS_CLIENT_init (struct GNUNET_SERVER_Handle *server)
+{
+ static struct GNUNET_SERVER_MessageHandler plugin_handlers[] = {
+ {&handle_dht_local_put, NULL,
+ GNUNET_MESSAGE_TYPE_DHT_CLIENT_PUT, 0},
+ {&handle_dht_local_get, NULL,
+ GNUNET_MESSAGE_TYPE_DHT_CLIENT_GET, 0},
+ {&handle_dht_local_get_stop, NULL,
+ GNUNET_MESSAGE_TYPE_DHT_CLIENT_GET_STOP,
+ sizeof (struct GNUNET_DHT_StopMessage) },
+ {NULL, NULL, 0, 0}
+ };
+ forward_map = GNUNET_CONTAINER_multihashmap_create (1024);
+ retry_heap = GNUNET_CONTAINER_heap_create (GNUNET_CONTAINER_HEAP_ORDER_MIN);
+ GNUNET_SERVER_add_handlers (server, plugin_handlers);
+ GNUNET_SERVER_disconnect_notify (server, &handle_client_disconnect, NULL);
+}
+
+
+/**
+ * Shutdown client subsystem.
+ */
+void
+GDS_CLIENT_done ()
+{
+ GNUNET_assert (client_head == NULL);
+ GNUNET_assert (client_tail == NULL);
+ if (GNUNET_SCHEDULER_NO_TASK != retry_task)
+ {
+ GNUNET_SCHEDULER_cancel (retry_task);
+ retry_task = GNUNET_SCHEDULER_NO_TASK;
+ }
+ GNUNET_assert (0 == GNUNET_CONTAINER_heap_get_size (retry_heap));
+ GNUNET_CONTAINER_heap_destroy (retry_heap);
+ retry_heap = NULL;
+ GNUNET_assert (0 == GNUNET_CONTAINER_multihashmap_size (forward_map));
+ GNUNET_CONTAINER_multihashmap_destroy (forward_map);
+ forward_map = NULL;
+}
+
+/* end of gnunet-service-dht_clients.c */
+
Added: gnunet/src/dht/gnunet-service-dht_clients.h
===================================================================
--- gnunet/src/dht/gnunet-service-dht_clients.h (rev 0)
+++ gnunet/src/dht/gnunet-service-dht_clients.h 2011-09-21 05:51:38 UTC (rev
16976)
@@ -0,0 +1,72 @@
+/*
+ This file is part of GNUnet.
+ (C) 2009, 2010, 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
+ 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
+ WITHOUT ANY WARRANTY; without even the implied warranty of
+ MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ General Public License for more details.
+
+ 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.
+*/
+
+/**
+ * @file dht/gnunet-service-dht_clients.h
+ * @brief GNUnet DHT service's client management code
+ * @author Christian Grothoff
+ * @author Nathan Evans
+ */
+#ifndef GNUNET_SERVICE_DHT_CLIENTS_H
+#define GNUNET_SERVICE_DHT_CLIENTS_H
+
+/**
+ * Handle a reply we've received from another peer. If the reply
+ * matches any of our pending queries, forward it to the respective
+ * client(s).
+ *
+ * @param expiration when will the reply expire
+ * @param key the query this reply is for
+ * @param get_path_length number of peers in 'get_path'
+ * @param get_path path the reply took on get
+ * @param put_path_length number of peers in 'put_path'
+ * @param put_path path the reply took on put
+ * @param type type of the reply
+ * @param data_size number of bytes in 'data'
+ * @param data application payload data
+ */
+void
+GDS_CLIENT_handle_reply (struct GNUNET_TIME_Absolute expiration,
+ const GNUNET_HashCode *key,
+ unsigned int get_path_length,
+ const struct GNUNET_PeerIdentity *get_path,
+ unsigned int put_path_length,
+ const struct GNUNET_PeerIdentity *put_path,
+ uint32_t type,
+ size_t data_size,
+ const void *data);
+
+
+/**
+ * Initialize client subsystem.
+ *
+ * @param server the initialized server
+ */
+void
+GDS_CLIENT_init (struct GNUNET_SERVER_Handle *server);
+
+
+/**
+ * Shutdown client subsystem.
+ */
+void
+GDS_CLIENT_done (void);
+
+#endif
Added: gnunet/src/dht/gnunet-service-dht_neighbours.c
===================================================================
--- gnunet/src/dht/gnunet-service-dht_neighbours.c
(rev 0)
+++ gnunet/src/dht/gnunet-service-dht_neighbours.c 2011-09-21 05:51:38 UTC
(rev 16976)
@@ -0,0 +1,353 @@
+/*
+ This file is part of GNUnet.
+ (C) 2009, 2010, 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
+ 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
+ WITHOUT ANY WARRANTY; without even the implied warranty of
+ MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ General Public License for more details.
+
+ 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.
+*/
+
+/**
+ * @file dht/gnunet-service-dht_neighbours.c
+ * @brief GNUnet DHT service's bucket and neighbour management code
+ * @author Christian Grothoff
+ * @author Nathan Evans
+ */
+
+#include "platform.h"
+#include "gnunet_block_lib.h"
+#include "gnunet_util_lib.h"
+#include "gnunet_protocols.h"
+#include "gnunet_nse_service.h"
+#include "gnunet_core_service.h"
+#include "gnunet_datacache_lib.h"
+#include "gnunet_transport_service.h"
+#include "gnunet_hello_lib.h"
+#include "gnunet_dht_service.h"
+#include "gnunet_statistics_service.h"
+#include "dht.h"
+#include <fenv.h>
+
+/**
+ * How many buckets will we allow total.
+ */
+#define MAX_BUCKETS sizeof (GNUNET_HashCode) * 8
+
+/**
+ * What is the maximum number of peers in a given bucket.
+ */
+#define DEFAULT_BUCKET_SIZE 4
+
+
+/**
+ * Linked list of messages to send to a particular other peer.
+ */
+struct P2PPendingMessage
+{
+ /**
+ * Pointer to next item in the list
+ */
+ struct P2PPendingMessage *next;
+
+ /**
+ * Pointer to previous item in the list
+ */
+ struct P2PPendingMessage *prev;
+
+ /**
+ * Message importance level.
+ */
+ unsigned int importance;
+
+ /**
+ * Time when this request was scheduled to be sent.
+ */
+ struct GNUNET_TIME_Absolute scheduled;
+
+ /**
+ * How long to wait before sending message.
+ */
+ struct GNUNET_TIME_Relative timeout;
+
+ /**
+ * Actual message to be sent, allocated at the end of the struct:
+ * // msg = (cast) &pm[1];
+ * // memcpy (&pm[1], data, len);
+ */
+ const struct GNUNET_MessageHeader *msg;
+
+};
+
+
+/**
+ * Entry for a peer in a bucket.
+ */
+struct PeerInfo
+{
+ /**
+ * Next peer entry (DLL)
+ */
+ struct PeerInfo *next;
+
+ /**
+ * Prev peer entry (DLL)
+ */
+ struct PeerInfo *prev;
+
+ /**
+ * Count of outstanding messages for peer.
+ */
+ unsigned int pending_count;
+
+ /**
+ * Head of pending messages to be sent to this peer.
+ */
+ struct P2PPendingMessage *head;
+
+ /**
+ * Tail of pending messages to be sent to this peer.
+ */
+ struct P2PPendingMessage *tail;
+
+ /**
+ * Core handle for sending messages to this peer.
+ */
+ struct GNUNET_CORE_TransmitHandle *th;
+
+ /**
+ * Preference update context
+ */
+ struct GNUNET_CORE_InformationRequestContext *info_ctx;
+
+ /**
+ * Task for scheduling message sends.
+ */
+ GNUNET_SCHEDULER_TaskIdentifier send_task;
+
+ /**
+ * Task for scheduling preference updates
+ */
+ GNUNET_SCHEDULER_TaskIdentifier preference_task;
+
+ /**
+ * What is the identity of the peer?
+ */
+ struct GNUNET_PeerIdentity id;
+
+#if 0
+ /**
+ * What is the average latency for replies received?
+ */
+ struct GNUNET_TIME_Relative latency;
+
+ /**
+ * Transport level distance to peer.
+ */
+ unsigned int distance;
+#endif
+
+};
+
+
+/**
+ * Peers are grouped into buckets.
+ */
+struct PeerBucket
+{
+ /**
+ * Head of DLL
+ */
+ struct PeerInfo *head;
+
+ /**
+ * Tail of DLL
+ */
+ struct PeerInfo *tail;
+
+ /**
+ * Number of peers in the bucket.
+ */
+ unsigned int peers_size;
+};
+
+
+/**
+ * The lowest currently used bucket.
+ */
+static unsigned int lowest_bucket; /* Initially equal to MAX_BUCKETS - 1
*/
+
+/**
+ * The buckets (Kademlia routing table, complete with growth).
+ * Array of size MAX_BUCKET_SIZE.
+ */
+static struct PeerBucket k_buckets[MAX_BUCKETS];
+
+/**
+ * Hash map of all known peers, for easy removal from k_buckets on disconnect.
+ */
+static struct GNUNET_CONTAINER_MultiHashMap *all_known_peers;
+
+/**
+ * Maximum size for each bucket.
+ */
+static unsigned int bucket_size = DEFAULT_BUCKET_SIZE;
+
+
+
+/**
+ * Method called whenever a peer connects.
+ *
+ * @param cls closure
+ * @param peer peer identity this notification is about
+ * @param atsi performance data
+ */
+static void
+handle_core_connect (void *cls, const struct GNUNET_PeerIdentity *peer,
+ const struct GNUNET_TRANSPORT_ATS_Information *atsi)
+{
+ struct PeerInfo *ret;
+ int peer_bucket;
+
+ /* Check for connect to self message */
+ if (0 == memcmp (&my_identity, peer, sizeof (struct GNUNET_PeerIdentity)))
+ return;
+
+#if DEBUG_DHT
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "%s:%s Receives core connect message for peer %s distance %d!\n",
+ my_short_id, "dht", GNUNET_i2s (peer), distance);
+#endif
+
+ if (GNUNET_YES ==
+ GNUNET_CONTAINER_multihashmap_contains (all_known_peers,
+ &peer->hashPubKey))
+ {
+#if DEBUG_DHT
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "%s:%s Received %s message for peer %s, but already have peer
in RT!",
+ my_short_id, "DHT", "CORE CONNECT", GNUNET_i2s (peer));
+#endif
+ GNUNET_break (0);
+ return;
+ }
+
+ peer_bucket = find_current_bucket (&peer->hashPubKey);
+ GNUNET_assert (peer_bucket >= lowest_bucket);
+ GNUNET_assert (peer_bucket < MAX_BUCKETS);
+ ret = GNUNET_malloc (sizeof (struct PeerInfo));
+#if 0
+ ret->latency = latency;
+ ret->distance = distance;
+#endif
+ ret->id = *peer;
+ GNUNET_CONTAINER_DLL_insert_after (k_buckets[peer_bucket].head,
+ k_buckets[peer_bucket].tail,
+ k_buckets[peer_bucket].tail, ret);
+ k_buckets[peer_bucket].peers_size++;
+ if ((GNUNET_CRYPTO_hash_matching_bits
+ (&my_identity.hashPubKey, &peer->hashPubKey) > 0) &&
+ (k_buckets[peer_bucket].peers_size <= bucket_size))
+ ret->preference_task =
+ GNUNET_SCHEDULER_add_now (&update_core_preference, ret);
+ if ((k_buckets[lowest_bucket].peers_size) >= bucket_size)
+ enable_next_bucket ();
+ newly_found_peers++;
+ GNUNET_CONTAINER_multihashmap_put (all_known_peers, &peer->hashPubKey, ret,
+
GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY);
+ increment_stats (STAT_PEERS_KNOWN);
+
+#if DEBUG_DHT
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "%s:%s Adding peer to routing list: %s\n", my_short_id, "DHT",
+ ret == NULL ? "NOT ADDED" : "PEER ADDED");
+#endif
+}
+
+
+/**
+ * Method called whenever a peer disconnects.
+ *
+ * @param cls closure
+ * @param peer peer identity this notification is about
+ */
+static void
+handle_core_disconnect (void *cls, const struct GNUNET_PeerIdentity *peer)
+{
+ struct PeerInfo *to_remove;
+ int current_bucket;
+
+ /* Check for disconnect from self message */
+ if (0 == memcmp (&my_identity, peer, sizeof (struct GNUNET_PeerIdentity)))
+ return;
+#if DEBUG_DHT
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "%s:%s: Received peer disconnect message for peer `%s' from
%s\n",
+ my_short_id, "DHT", GNUNET_i2s (peer), "CORE");
+#endif
+
+ if (GNUNET_YES !=
+ GNUNET_CONTAINER_multihashmap_contains (all_known_peers,
+ &peer->hashPubKey))
+ {
+ GNUNET_break (0);
+#if DEBUG_DHT
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "%s:%s: do not have peer `%s' in RT, can't disconnect!\n",
+ my_short_id, "DHT", GNUNET_i2s (peer));
+#endif
+ return;
+ }
+ increment_stats (STAT_DISCONNECTS);
+ GNUNET_assert (GNUNET_CONTAINER_multihashmap_contains
+ (all_known_peers, &peer->hashPubKey));
+ to_remove =
+ GNUNET_CONTAINER_multihashmap_get (all_known_peers, &peer->hashPubKey);
+ GNUNET_assert (to_remove != NULL);
+ if (NULL != to_remove->info_ctx)
+ {
+ GNUNET_CORE_peer_change_preference_cancel (to_remove->info_ctx);
+ to_remove->info_ctx = NULL;
+ }
+ GNUNET_assert (0 ==
+ memcmp (peer, &to_remove->id,
+ sizeof (struct GNUNET_PeerIdentity)));
+ current_bucket = find_current_bucket (&to_remove->id.hashPubKey);
+ delete_peer (to_remove, current_bucket);
+}
+
+
+
+/**
+ * Initialize neighbours subsystem.
+ */
+void
+GST_NEIGHBOURS_init ()
+{
+}
+
+
+/**
+ * Shutdown neighbours subsystem.
+ */
+void
+GST_NEIGHBOURS_done ()
+{
+}
+
+
+
+
+
+
+/* end of gnunet-service-dht_neighbours.c */
Added: gnunet/src/dht/gnunet-service-dht_neighbours.h
===================================================================
--- gnunet/src/dht/gnunet-service-dht_neighbours.h
(rev 0)
+++ gnunet/src/dht/gnunet-service-dht_neighbours.h 2011-09-21 05:51:38 UTC
(rev 16976)
@@ -0,0 +1,122 @@
+/*
+ This file is part of GNUnet.
+ (C) 2009, 2010, 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
+ 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
+ WITHOUT ANY WARRANTY; without even the implied warranty of
+ MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ General Public License for more details.
+
+ 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.
+*/
+
+/**
+ * @file dht/gnunet-service-dht_neighbours.h
+ * @brief GNUnet DHT routing code
+ * @author Christian Grothoff
+ * @author Nathan Evans
+ */
+#ifndef GNUNET_SERVICE_DHT_NEIGHBOURS_H
+#define GNUNET_SERVICE_DHT_NEIGHBOURS_H
+
+
+/**
+ * Perform a PUT operation.
+ *
+ * @param type type of the block
+ * @param options routing options
+ * @param desired_replication_level desired replication count
+ * @param expiration_time when does the content expire
+ * @param key key for the content
+ * @param put_path_length number of entries in put_path
+ * @param put_path peers this request has traversed so far (if tracked)
+ * @param data payload to store
+ * @param data_size number of bytes in data
+ */
+void
+GST_NEIGHBOURS_handle_put (uint32_t type,
+ uint32_t options,
+ uint32_t desired_replication_level,
+ GNUNET_TIME_Absolute expiration_time,
+ const GNUNET_HashCode *key,
+ unsigned int put_path_length,
+ struct GNUNET_PeerIdentity *put_path,
+ const void *data,
+ size_t data_size);
+
+
+/**
+ * Perform a GET operation.
+ *
+ *
+ * @param type type of the block
+ * @param options routing options
+ * @param desired_replication_level desired replication count
+ * @param key key for the content
+ * @param xquery extended query
+ * @param xquery_size number of bytes in xquery
+ * @param reply_bf bloomfilter to filter duplicates
+ * @param reply_bf_mutator mutator for reply_bf
+ * @param peer_bf filter for peers not to select (again)
+ */
+void
+GST_NEIGHBOURS_handle_get (uint32_t type,
+ uint32_t options,
+ uint32_t desired_replication_level,
+ const GNUNET_HashCode *key,
+ const void *xquery,
+ size_t xquery_size,
+ const struct GNUNET_CONTAINER_BloomFilter *reply_bf,
+ uint32_t reply_bf_mutator,
+ const struct GNUNET_CONTAINER_BloomFilter *peer_bf);
+
+
+/**
+ * Handle a reply (route to origin).
+ *
+ * @param type type of the block
+ * @param options routing options
+ * @param expiration_time when does the content expire
+ * @param key key for the content
+ * @param put_path_length number of entries in put_path
+ * @param put_path peers the original PUT traversed (if tracked)
+ * @param get_path_length number of entries in put_path
+ * @param get_path peers this reply has traversed so far (if tracked)
+ * @param data payload of the reply
+ * @param data_size number of bytes in data
+ */
+void
+GST_NEIGHBOURS_handle_reply (uint32_t type,
+ uint32_t options,
+ GNUNET_TIME_Absolute expiration_time,
+ const GNUNET_HashCode *key,
+ unsigned int put_path_length,
+ struct GNUNET_PeerIdentity *put_path,
+ unsigned int get_path_length,
+ struct GNUNET_PeerIdentity *get_path,
+ const void *data,
+ size_t data_size);
+
+
+/**
+ * Initialize neighbours subsystem.
+ */
+void
+GST_NEIGHBOURS_init (void);
+
+/**
+ * Shutdown neighbours subsystem.
+ */
+void
+GST_NEIGHBOURS_done (void);
+
+
+#endif
Added: gnunet/src/dht/gnunet-service-dht_nse.c
===================================================================
--- gnunet/src/dht/gnunet-service-dht_nse.c (rev 0)
+++ gnunet/src/dht/gnunet-service-dht_nse.c 2011-09-21 05:51:38 UTC (rev
16976)
@@ -0,0 +1,84 @@
+/*
+ This file is part of GNUnet.
+ (C) 2009, 2010, 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
+ 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
+ WITHOUT ANY WARRANTY; without even the implied warranty of
+ MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ General Public License for more details.
+
+ 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.
+*/
+
+/**
+ * @file dht/gnunet-service-dht_nse.c
+ * @brief GNUnet DHT integration with NSE
+ * @author Christian Grothoff
+ */
+#include "gnunet-service-dht.h"
+#include "gnunet-service-dht_nse.h"
+
+/**
+ * log of the current network size estimate, used as the point where
+ * we switch between random and deterministic routing. Default
+ * value of 4.0 is used if NSE module is not available (i.e. not
+ * configured).
+ */
+static double log_of_network_size_estimate = 4.0;
+
+/**
+ * Network size estimation handle.
+ */
+static struct GNUNET_NSE_Handle *nse;
+
+
+/**
+ * Callback that is called when network size estimate is updated.
+ *
+ * @param cls closure
+ * @param timestamp time when the estimate was received from the server (or
created by the server)
+ * @param logestimate the log(Base 2) value of the current network size
estimate
+ * @param std_dev standard deviation for the estimate
+ *
+ */
+static void
+update_network_size_estimate (void *cls, struct GNUNET_TIME_Absolute timestamp,
+ double logestimate, double std_dev)
+{
+ log_of_network_size_estimate = logestimate;
+}
+
+
+double
+GDS_nse_get ()
+{
+ return log_of_network_size_estimate;
+}
+
+
+void
+GDS_nse_init ()
+{
+ nse = GNUNET_NSE_connect (GDS_cfg, &update_network_size_estimate, NULL);
+}
+
+
+void
+GDS_nse_done ()
+{
+ if (NULL != nse)
+ {
+ GNUNET_NSE_disconnect (nse);
+ nse = NULL;
+ }
+}
+
+/* end of gnunet-service-dht_nse.c */
Added: gnunet/src/dht/gnunet-service-dht_nse.h
===================================================================
--- gnunet/src/dht/gnunet-service-dht_nse.h (rev 0)
+++ gnunet/src/dht/gnunet-service-dht_nse.h 2011-09-21 05:51:38 UTC (rev
16976)
@@ -0,0 +1,40 @@
+/*
+ This file is part of GNUnet.
+ (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
+ 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
+ WITHOUT ANY WARRANTY; without even the implied warranty of
+ MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ General Public License for more details.
+
+ 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.
+*/
+
+/**
+ * @file dht/gnunet-service-dht_nse.h
+ * @brief GNUnet DHT integration with NSE
+ * @author Christian Grothoff
+ */
+#ifndef GNUNET_SERVICE_DHT_NSE_H
+#define GNUNET_SERVICE_DHT_NSE_H
+
+
+double
+GDS_nse_get (void);
+
+
+void
+GDS_nse_init (void);
+
+void
+GDS_nse_done (void);
+
+#endif
| [Prev in Thread] |
Current Thread |
[Next in Thread] |
- [GNUnet-SVN] r16976 - gnunet/src/dht,
gnunet <=