#include "gnunet_dht_service.h"
#include "dht.h"
-#define DEBUG_DHT_API GNUNET_NO
+#define DEBUG_DHT_API GNUNET_EXTRA_LOGGING
+
+#define LOG(kind,...) GNUNET_log_from (kind, "dht-api",__VA_ARGS__)
/**
* Entry in our list of messages to be (re-)transmitted.
};
+/**
+ * Handle to a monitoring request.
+ */
+struct GNUNET_DHT_MonitorHandle
+{
+ /**
+ * DLL.
+ */
+ struct GNUNET_DHT_MonitorHandle *next;
+
+ /**
+ * DLL.
+ */
+ struct GNUNET_DHT_MonitorHandle *prev;
+
+ /**
+ * Main handle to this DHT api.
+ */
+ struct GNUNET_DHT_Handle *dht_handle;
+
+ /**
+ * Type of block looked for.
+ */
+ enum GNUNET_BLOCK_Type type;
+
+ /**
+ * Key being looked for, NULL == all.
+ */
+ GNUNET_HashCode *key;
+
+ /**
+ * Callback for each received message of interest.
+ */
+ GNUNET_DHT_MonitorCB cb;
+
+ /**
+ * Closure for cb.
+ */
+ void *cb_cls;
+
+};
+
+
/**
* Connection to the DHT service.
*/
*/
struct PendingMessage *pending_tail;
+ /**
+ * Head of linked list of messages we would like to monitor.
+ */
+ struct GNUNET_DHT_MonitorHandle *monitor_head;
+
+ /**
+ * Tail of linked list of messages we would like to monitor.
+ */
+ struct GNUNET_DHT_MonitorHandle *monitor_tail;
+
/**
* Hash map containing the current outstanding unique requests
* (values are of type 'struct GNUNET_DHT_RouteHandle').
*/
uint64_t uid_gen;
+ /**
+ * Did we start our receive loop yet?
+ */
+ int in_receive;
};
{
if (handle->client != NULL)
return GNUNET_OK;
+ handle->in_receive = GNUNET_NO;
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"));
+ 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;
}
if (GNUNET_NO == rh->message->in_pending_queue)
{
+#if DEBUG_DHT
+ LOG (GNUNET_ERROR_TYPE_DEBUG,
+ "Retransmitting request related to %s to DHT %p\n", GNUNET_h2s (key),
+ handle);
+#endif
GNUNET_CONTAINER_DLL_insert (handle->pending_head, handle->pending_tail,
rh->message);
rh->message->in_pending_queue = GNUNET_YES;
{
struct GNUNET_DHT_Handle *handle = cls;
+#if DEBUG_DHT
+ LOG (GNUNET_ERROR_TYPE_DEBUG, "Reconnecting with DHT %p\n", handle);
+#endif
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;
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)
+ if (GNUNET_YES != try_connect (handle))
{
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "dht reconnect failed(!)\n");
+#if DEBUG_DHT
+ LOG (GNUNET_ERROR_TYPE_DEBUG, "dht reconnect failed(!)\n");
+#endif
return;
}
GNUNET_CONTAINER_multihashmap_iterate (handle->active_requests,
return;
GNUNET_assert (handle->reconnect_task == GNUNET_SCHEDULER_NO_TASK);
if (NULL != handle->th)
- GNUNET_CLIENT_notify_transmit_ready_cancel(handle->th);
+ GNUNET_CLIENT_notify_transmit_ready_cancel (handle->th);
handle->th = NULL;
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "Disconnecting from DHT service, will try to reconnect in %llu ms\n",
+ (unsigned long long) handle->retry_time.rel_value);
GNUNET_CLIENT_disconnect (handle->client, GNUNET_NO);
handle->client = NULL;
handle->reconnect_task =
if (handle->client == NULL)
{
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "process_pending_messages called, but client is null, reconnecting\n");
do_disconnect (handle);
return;
}
handle);
if (NULL != handle->th)
return;
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "notify_transmit_ready returned NULL, reconnecting\n");
do_disconnect (handle);
}
handle->th = NULL;
if (buf == NULL)
{
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Transmission to DHT service failed! Reconnecting!\n");
+#if DEBUG_DHT
+ LOG (GNUNET_ERROR_TYPE_DEBUG,
+ "Transmission to DHT service failed! Reconnecting!\n");
+#endif
do_disconnect (handle);
return 0;
}
memcpy (buf, head->msg, tsize);
GNUNET_CONTAINER_DLL_remove (handle->pending_head, handle->pending_tail,
head);
+ head->in_pending_queue = GNUNET_NO;
if (head->timeout_task != GNUNET_SCHEDULER_NO_TASK)
{
GNUNET_SCHEDULER_cancel (head->timeout_task);
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);
+#if DEBUG_DHT
+ LOG (GNUNET_ERROR_TYPE_DEBUG,
+ "Forwarded request of %u bytes to DHT service\n", (unsigned int) tsize);
+#endif
+ if (GNUNET_NO == handle->in_receive)
+ {
+#if DEBUG_DHT
+ LOG (GNUNET_ERROR_TYPE_DEBUG, "Starting to process replies from DHT\n");
+#endif
+ handle->in_receive = GNUNET_YES;
+ GNUNET_CLIENT_receive (handle->client, &service_message_handler, handle,
+ GNUNET_TIME_UNIT_FOREVER_REL);
+ }
return tsize;
}
if (dht_msg->unique_id != get_handle->unique_id)
{
/* UID mismatch */
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Ignoring reply (UID mismatch: %llu/%llu)\n",
- dht_msg->unique_id,
- get_handle->unique_id);
+#if DEBUG_DHT
+ LOG (GNUNET_ERROR_TYPE_DEBUG,
+ "Ignoring reply for %s: UID mismatch: %llu/%llu\n", GNUNET_h2s (key),
+ dht_msg->unique_id, get_handle->unique_id);
+#endif
return GNUNET_YES;
}
msize = ntohs (dht_msg->header.size);
put_path_length = ntohl (dht_msg->put_path_length);
get_path_length = ntohl (dht_msg->get_path_length);
- meta_length = sizeof (struct GNUNET_DHT_ClientResultMessage) +
- sizeof (struct GNUNET_PeerIdentity) * (get_path_length + put_path_length);
- if ( (msize < meta_length) ||
- (get_path_length > GNUNET_SERVER_MAX_MESSAGE_SIZE / sizeof (struct GNUNET_PeerIdentity)) ||
- (put_path_length > GNUNET_SERVER_MAX_MESSAGE_SIZE / sizeof (struct GNUNET_PeerIdentity)) )
- {
- GNUNET_break (0);
- return GNUNET_NO;
- }
- data_length = msize - meta_length;
+ meta_length =
+ sizeof (struct GNUNET_DHT_ClientResultMessage) +
+ sizeof (struct GNUNET_PeerIdentity) * (get_path_length + put_path_length);
+ if ((msize < meta_length) ||
+ (get_path_length >
+ GNUNET_SERVER_MAX_MESSAGE_SIZE / sizeof (struct GNUNET_PeerIdentity)) ||
+ (put_path_length >
+ GNUNET_SERVER_MAX_MESSAGE_SIZE / sizeof (struct GNUNET_PeerIdentity)))
+ {
+ GNUNET_break (0);
+ return GNUNET_NO;
+ }
+ data_length = msize - meta_length;
+#if DEBUG_DHT
+ LOG (GNUNET_ERROR_TYPE_DEBUG, "Giving %u byte reply for %s to application\n",
+ (unsigned int) data_length, GNUNET_h2s (key));
+#endif
put_path = (const struct GNUNET_PeerIdentity *) &dht_msg[1];
get_path = &put_path[put_path_length];
data = &get_path[get_path_length];
get_handle->iter (get_handle->iter_cls,
- GNUNET_TIME_absolute_ntoh (dht_msg->expiration),
- key,
- get_path, get_path_length,
- put_path, put_path_length,
- ntohl (dht_msg->type),
- data_length, data);
+ GNUNET_TIME_absolute_ntoh (dht_msg->expiration), key,
+ get_path, get_path_length, put_path, put_path_length,
+ ntohl (dht_msg->type), data_length, data);
return GNUNET_YES;
}
+/**
+ * Process a monitoring message from the service.
+ *
+ * @param handle The DHT handle.
+ * @param msg Message from the service.
+ *
+ * @return GNUNET_OK if everything went fine,
+ * GNUNET_SYSERR if the message is malformed.
+ */
+static int
+process_monitor_message (struct GNUNET_DHT_Handle *handle,
+ const struct GNUNET_MessageHeader *msg)
+{
+ struct GNUNET_DHT_MonitorMessage *m;
+ struct GNUNET_DHT_MonitorHandle *h;
+ size_t msize;
+
+ if (ntohs (msg->type) < GNUNET_MESSAGE_TYPE_DHT_MONITOR_GET ||
+ ntohs (msg->type) > GNUNET_MESSAGE_TYPE_DHT_MONITOR_PUT)
+ return GNUNET_SYSERR;
+ msize = ntohs (msg->size);
+ if (msize < sizeof (struct GNUNET_DHT_MonitorMessage))
+ return GNUNET_SYSERR;
+
+ m = (struct GNUNET_DHT_MonitorMessage *) msg;
+ h = handle->monitor_head;
+ while (NULL != h)
+ {
+ if (h->type == ntohl(m->type) &&
+ (NULL == h->key ||
+ memcmp (h->key, &m->key, sizeof (GNUNET_HashCode)) == 0))
+ {
+ struct GNUNET_PeerIdentity *path;
+ uint32_t getl;
+ uint32_t putl;
+
+ path = (struct GNUNET_PeerIdentity *) &m[1];
+ getl = ntohl (m->get_path_length);
+ putl = ntohl (m->put_path_length);
+ h->cb (h->cb_cls, ntohs(msg->type),
+ GNUNET_TIME_absolute_ntoh(m->expiration),
+ &m->key,
+ &path[getl], putl, path, getl,
+ ntohl (m->desired_replication_level),
+ ntohl (m->options), ntohl (m->type),
+ (void *) &path[getl + putl],
+ ntohs (msg->size) -
+ sizeof (struct GNUNET_DHT_MonitorMessage) -
+ sizeof (struct GNUNET_PeerIdentity) * (putl + getl));
+ }
+ h = h->next;
+ }
+
+ return GNUNET_OK;
+}
+
/**
* Handler for messages received from the DHT service
* a demultiplexer which handles numerous message types
if (msg == NULL)
{
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Error receiving data from DHT service, reconnecting\n");
+#if DEBUG_DHT
+ LOG (GNUNET_ERROR_TYPE_DEBUG,
+ "Error receiving data from DHT service, reconnecting\n");
+#endif
do_disconnect (handle);
return;
}
if (ntohs (msg->type) != GNUNET_MESSAGE_TYPE_DHT_CLIENT_RESULT)
{
+ if (process_monitor_message (handle, msg) == GNUNET_OK)
+ {
+ GNUNET_CLIENT_receive (handle->client, &service_message_handler, handle,
+ GNUNET_TIME_UNIT_FOREVER_REL);
+ return;
+ }
GNUNET_break (0);
do_disconnect (handle);
return;
do_disconnect (handle);
return;
}
- GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Received reply from DHT service\n");
dht_msg = (const struct GNUNET_DHT_ClientResultMessage *) msg;
+#if DEBUG_DHT
+ LOG (GNUNET_ERROR_TYPE_DEBUG, "Received reply for `%s' from DHT service %p\n",
+ GNUNET_h2s (&dht_msg->key), handle);
+#endif
GNUNET_CONTAINER_multihashmap_get_multiple (handle->active_requests,
&dht_msg->key, &process_reply,
(void *) dht_msg);
}
while (NULL != (pm = handle->pending_head))
{
+ GNUNET_assert (GNUNET_YES == pm->in_pending_queue);
GNUNET_CONTAINER_DLL_remove (handle->pending_head, handle->pending_tail,
pm);
+ pm->in_pending_queue = GNUNET_NO;
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)
struct GNUNET_DHT_Handle *handle;
handle = pending->handle;
+ GNUNET_assert (GNUNET_YES == pending->in_pending_queue);
GNUNET_CONTAINER_DLL_remove (handle->pending_head, handle->pending_tail,
pending);
+ pending->in_pending_queue = GNUNET_NO;
if (pending->cont != NULL)
pending->cont (pending->cont_cls, tc);
GNUNET_free (pending);
struct PendingMessage *pending;
msize = sizeof (struct GNUNET_DHT_ClientPutMessage) + size;
- if ( (msize >= GNUNET_SERVER_MAX_MESSAGE_SIZE) ||
- (size >= GNUNET_SERVER_MAX_MESSAGE_SIZE) )
- {
- GNUNET_break (0);
- if (NULL != cont)
- cont (cont_cls, NULL);
- return;
- }
+ if ((msize >= GNUNET_SERVER_MAX_MESSAGE_SIZE) ||
+ (size >= GNUNET_SERVER_MAX_MESSAGE_SIZE))
+ {
+ GNUNET_break (0);
+ if (NULL != cont)
+ cont (cont_cls, NULL);
+ return;
+ }
pending = GNUNET_malloc (sizeof (struct PendingMessage) + msize);
put_msg = (struct GNUNET_DHT_ClientPutMessage *) &pending[1];
pending->msg = &put_msg->header;
pending->cont_cls = cont_cls;
pending->free_on_send = GNUNET_YES;
pending->timeout_task =
- GNUNET_SCHEDULER_add_delayed (timeout, &timeout_put_request, pending);
+ GNUNET_SCHEDULER_add_delayed (timeout, &timeout_put_request, pending);
put_msg->header.size = htons (msize);
put_msg->header.type = htons (GNUNET_MESSAGE_TYPE_DHT_CLIENT_PUT);
put_msg->type = htonl (type);
struct GNUNET_TIME_Relative timeout,
enum GNUNET_BLOCK_Type type, const GNUNET_HashCode * key,
uint32_t desired_replication_level,
- enum GNUNET_DHT_RouteOption options,
- const void *xquery, size_t xquery_size,
- GNUNET_DHT_GetIterator iter, void *iter_cls)
+ enum GNUNET_DHT_RouteOption options, const void *xquery,
+ size_t xquery_size, GNUNET_DHT_GetIterator iter,
+ void *iter_cls)
{
struct GNUNET_DHT_ClientGetMessage *get_msg;
struct GNUNET_DHT_GetHandle *get_handle;
struct PendingMessage *pending;
msize = sizeof (struct GNUNET_DHT_ClientGetMessage) + xquery_size;
- if ( (msize >= GNUNET_SERVER_MAX_MESSAGE_SIZE) ||
- (xquery_size >= GNUNET_SERVER_MAX_MESSAGE_SIZE) )
- {
- GNUNET_break (0);
- return NULL;
- }
+ if ((msize >= GNUNET_SERVER_MAX_MESSAGE_SIZE) ||
+ (xquery_size >= GNUNET_SERVER_MAX_MESSAGE_SIZE))
+ {
+ GNUNET_break (0);
+ return NULL;
+ }
+#if DEBUG_DHT
+ LOG (GNUNET_ERROR_TYPE_DEBUG, "Sending query for %s to DHT %p\n",
+ GNUNET_h2s (key), handle);
+#endif
pending = GNUNET_malloc (sizeof (struct PendingMessage) + msize);
get_msg = (struct GNUNET_DHT_ClientGetMessage *) &pending[1];
pending->msg = &get_msg->header;
get_handle = GNUNET_malloc (sizeof (struct GNUNET_DHT_GetHandle));
get_handle->iter = iter;
get_handle->iter_cls = iter_cls;
- get_handle->message = pending;
+ get_handle->message = pending;
get_handle->unique_id = get_msg->unique_id;
- GNUNET_CONTAINER_multihashmap_put (handle->active_requests,
- key, get_handle,
- GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
+ GNUNET_CONTAINER_multihashmap_put (handle->active_requests, key, get_handle,
+ GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
process_pending_messages (handle);
return get_handle;
}
struct PendingMessage *pending;
handle = get_handle->message->handle;
- get_msg = (const struct GNUNET_DHT_ClientGetMessage*) get_handle->message->msg;
-
+ get_msg =
+ (const struct GNUNET_DHT_ClientGetMessage *) get_handle->message->msg;
+#if DEBUG_DHT
+ LOG (GNUNET_ERROR_TYPE_DEBUG, "Sending STOP for %s to DHT via %p\n",
+ GNUNET_h2s (&get_msg->key), handle);
+#endif
/* generate STOP */
- pending = GNUNET_malloc (sizeof (struct PendingMessage) + sizeof (struct GNUNET_DHT_ClientGetStopMessage));
+ pending =
+ GNUNET_malloc (sizeof (struct PendingMessage) +
+ sizeof (struct GNUNET_DHT_ClientGetStopMessage));
stop_msg = (struct GNUNET_DHT_ClientGetStopMessage *) &pending[1];
pending->msg = &stop_msg->header;
pending->handle = handle;
pending->free_on_send = GNUNET_YES;
- stop_msg->header.size = htons (sizeof (struct GNUNET_DHT_ClientGetStopMessage));
+ stop_msg->header.size =
+ htons (sizeof (struct GNUNET_DHT_ClientGetStopMessage));
stop_msg->header.type = htons (GNUNET_MESSAGE_TYPE_DHT_CLIENT_GET_STOP);
stop_msg->reserved = htonl (0);
stop_msg->unique_id = get_msg->unique_id;
/* remove 'GET' from active status */
GNUNET_assert (GNUNET_YES ==
- GNUNET_CONTAINER_multihashmap_remove (handle->active_requests,
- &get_msg->key, get_handle));
+ GNUNET_CONTAINER_multihashmap_remove (handle->active_requests,
+ &get_msg->key,
+ get_handle));
if (GNUNET_YES == get_handle->message->in_pending_queue)
- {
- GNUNET_CONTAINER_DLL_remove (handle->pending_head,
- handle->pending_tail,
- get_handle->message);
- get_handle->message->in_pending_queue = GNUNET_NO;
- }
+ {
+ GNUNET_CONTAINER_DLL_remove (handle->pending_head, handle->pending_tail,
+ get_handle->message);
+ get_handle->message->in_pending_queue = GNUNET_NO;
+ }
GNUNET_free (get_handle->message);
GNUNET_free (get_handle);
}
+/**
+ * Start monitoring the local DHT service.
+ *
+ * @param handle Handle to the DHT service.
+ * @param type Type of blocks that are of interest.
+ * @param key Key of data of interest, NULL for all.
+ * @param cb Callback to process all monitored data.
+ * @param cb_cls Closure for cb.
+ *
+ * @return Handle to stop monitoring.
+ */
+struct GNUNET_DHT_MonitorHandle *
+GNUNET_DHT_monitor_start (struct GNUNET_DHT_Handle *handle,
+ enum GNUNET_BLOCK_Type type,
+ const GNUNET_HashCode *key,
+ GNUNET_DHT_MonitorCB cb,
+ void *cb_cls)
+{
+ struct GNUNET_DHT_MonitorHandle *h;
+ struct GNUNET_DHT_MonitorMessage *m;
+ struct PendingMessage *pending;
+
+ h = GNUNET_malloc (sizeof (struct GNUNET_DHT_MonitorHandle));
+ GNUNET_CONTAINER_DLL_insert(handle->monitor_head, handle->monitor_tail, h);
+
+ GNUNET_assert (NULL != cb);
+ h->cb = cb;
+ h->cb_cls = cb_cls;
+ h->type = type;
+ h->dht_handle = handle;
+ if (NULL != key)
+ {
+ h->key = GNUNET_malloc (sizeof(GNUNET_HashCode));
+ memcpy (h->key, key, sizeof(GNUNET_HashCode));
+ }
+
+ pending = GNUNET_malloc (sizeof (struct GNUNET_DHT_MonitorMessage) +
+ sizeof (struct PendingMessage));
+ m = (struct GNUNET_DHT_MonitorMessage *) &pending[1];
+ pending->msg = &m->header;
+ pending->handle = handle;
+ pending->free_on_send = GNUNET_YES;
+ m->header.type = htons (GNUNET_MESSAGE_TYPE_DHT_MONITOR_GET);
+ m->header.size = htons (sizeof (struct GNUNET_DHT_MonitorMessage));
+ m->type = htonl(type);
+ if (NULL != key)
+ memcpy (&m->key, key, sizeof(GNUNET_HashCode));
+ GNUNET_CONTAINER_DLL_insert (handle->pending_head, handle->pending_tail,
+ pending);
+ pending->in_pending_queue = GNUNET_YES;
+ process_pending_messages (handle);
+
+ return h;
+}
+
+
+/**
+ * Stop monitoring.
+ *
+ * @param handle The handle to the monitor request returned by monitor_start.
+ *
+ * On return get_handle will no longer be valid, caller must not use again!!!
+ */
+void
+GNUNET_DHT_monitor_stop (struct GNUNET_DHT_MonitorHandle *handle)
+{
+ GNUNET_free_non_null (handle->key);
+ GNUNET_CONTAINER_DLL_remove (handle->dht_handle->monitor_head,
+ handle->dht_handle->monitor_tail,
+ handle);
+ GNUNET_free (handle);
+}
+
+
+
/* end of dht_api.c */