/* 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 2, 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 main DHT service shell, building block for DHT implementations * @author Christian Grothoff * @author Nathan Evans */ #include "platform.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_core_service.h" #include "gnunet_signal_lib.h" #include "gnunet_util_lib.h" #include "gnunet_datacache_lib.h" #include "dht.h" /** * Handle to the datacache service (for inserting/retrieving data) */ struct GNUNET_DATACACHE_Handle *datacache; /** * The main scheduler to use for the DHT service */ static struct GNUNET_SCHEDULER_Handle *sched; /** * The configuration the DHT service is running with */ static const struct GNUNET_CONFIGURATION_Handle *cfg; /** * Timeout for transmissions to clients */ static struct GNUNET_TIME_Relative client_transmit_timeout; /** * Handle to the core service */ static struct GNUNET_CORE_Handle *coreAPI; /** * The identity of our peer. */ static struct GNUNET_PeerIdentity my_identity; /** * Task to run when we shut down, cleaning up all our trash */ static GNUNET_SCHEDULER_TaskIdentifier cleanup_task; /** * Context for handling results from a get request. */ struct DatacacheGetContext { /** * The client to send the result to. */ struct GNUNET_SERVER_Client *client; /** * The unique id of this request */ unsigned long long unique_id; }; struct DHT_MessageContext { /** * The client this request was received from. */ struct GNUNET_SERVER_Client *client; /** * The key this request was about */ GNUNET_HashCode *key; /** * The unique identifier of this request */ unsigned long long unique_id; /** * Desired replication level */ size_t replication; /** * Any message options for this request */ size_t msg_options; }; /** * Server handler for handling locally received dht requests */ static void handle_dht_start_message (void *cls, struct GNUNET_SERVER_Client *client, const struct GNUNET_MessageHeader *message); static void handle_dht_stop_message (void *cls, struct GNUNET_SERVER_Client *client, const struct GNUNET_MessageHeader *message); static struct GNUNET_SERVER_MessageHandler plugin_handlers[] = { {&handle_dht_start_message, NULL, GNUNET_MESSAGE_TYPE_DHT, 0}, {&handle_dht_stop_message, NULL, GNUNET_MESSAGE_TYPE_DHT_STOP, 0}, {NULL, NULL, 0, 0} }; /** * Core handler for p2p dht get requests. */ static int handle_dht_p2p_get (void *cls, const struct GNUNET_PeerIdentity *peer, const struct GNUNET_MessageHeader *message, struct GNUNET_TIME_Relative latency, uint32_t distance); /** * Core handler for p2p dht put requests. */ static int handle_dht_p2p_put (void *cls, const struct GNUNET_PeerIdentity *peer, const struct GNUNET_MessageHeader *message, struct GNUNET_TIME_Relative latency, uint32_t distance); /** * Core handler for p2p dht find peer requests. */ static int handle_dht_p2p_find_peer (void *cls, const struct GNUNET_PeerIdentity *peer, const struct GNUNET_MessageHeader *message, struct GNUNET_TIME_Relative latency, uint32_t distance); static struct GNUNET_CORE_MessageHandler core_handlers[] = { {&handle_dht_p2p_get, GNUNET_MESSAGE_TYPE_DHT_GET, 0}, {&handle_dht_p2p_put, GNUNET_MESSAGE_TYPE_DHT_PUT, 0}, {&handle_dht_p2p_find_peer, GNUNET_MESSAGE_TYPE_DHT_FIND_PEER, 0}, {NULL, 0, 0} }; static size_t send_reply (void *cls, size_t size, void *buf) { struct GNUNET_DHT_Message *reply = cls; if (buf == NULL) /* Message timed out, that's crappy... */ { GNUNET_free (reply); return 0; } if (size >= ntohs (reply->header.size)) { memcpy (buf, reply, ntohs (reply->header.size)); return ntohs (reply->header.size); } else return 0; } static void send_reply_to_client (struct GNUNET_SERVER_Client *client, struct GNUNET_MessageHeader *message, unsigned long long uid) { struct GNUNET_DHT_Message *reply; size_t msize; size_t tsize; #if DEBUG_DHT GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "`%s': Sending reply to client.\n", "DHT"); #endif msize = ntohs (message->size); tsize = sizeof (struct GNUNET_DHT_Message) + msize; reply = GNUNET_malloc (tsize); reply->header.type = htons (GNUNET_MESSAGE_TYPE_DHT); reply->header.size = htons (tsize); if (uid != 0) reply->unique = htons (GNUNET_YES); reply->unique_id = GNUNET_htonll (uid); memcpy (&reply[1], message, msize); GNUNET_SERVER_notify_transmit_ready (client, tsize, GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 5), &send_reply, reply); } /** * Iterator for local get request results, 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, uint32_t size, const char *data, uint32_t type) { struct DatacacheGetContext *datacache_get_ctx = cls; struct GNUNET_DHT_GetResultMessage *get_result; get_result = GNUNET_malloc (sizeof (struct GNUNET_DHT_GetResultMessage) + size); get_result->header.type = htons (GNUNET_MESSAGE_TYPE_DHT_GET_RESULT); get_result->header.size = htons (sizeof (struct GNUNET_DHT_GetResultMessage) + size); get_result->data_size = htons (size); get_result->expiration = exp; memcpy (&get_result->key, key, sizeof (GNUNET_HashCode)); get_result->type = htons (type); memcpy (&get_result[1], data, size); send_reply_to_client (datacache_get_ctx->client, &get_result->header, datacache_get_ctx->unique_id); GNUNET_free (get_result); return GNUNET_OK; } /** * Server handler for initiating local dht get requests */ static void handle_dht_get (void *cls, struct GNUNET_DHT_GetMessage *get_msg, struct DHT_MessageContext *message_context) { #if DEBUG_DHT GNUNET_HashCode get_key; #endif size_t get_type; unsigned int results; struct DatacacheGetContext *datacache_get_context; GNUNET_assert (ntohs (get_msg->header.size) >= sizeof (struct GNUNET_DHT_GetMessage)); get_type = ntohs (get_msg->type); #if DEBUG_DHT GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "`%s': Received `%s' request from client, message type %d, key %s, uid %llu\n", "DHT", "GET", get_type, GNUNET_h2s (&get_key), message_context->unique_id); #endif datacache_get_context = GNUNET_malloc (sizeof (struct DatacacheGetContext)); datacache_get_context->client = message_context->client; datacache_get_context->unique_id = message_context->unique_id; results = 0; if (datacache != NULL) results = GNUNET_DATACACHE_get (datacache, message_context->key, get_type, datacache_get_iterator, datacache_get_context); #if DEBUG_DHT GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "`%s': Found %d results for local `%s' request\n", "DHT", results, "GET"); #endif GNUNET_free (datacache_get_context); /* FIXME: Implement get functionality here */ } /** * Server handler for initiating local dht find peer requests */ static void handle_dht_find_peer (void *cls, struct GNUNET_DHT_FindPeerMessage *find_msg, struct DHT_MessageContext *message_context) { #if DEBUG_DHT GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "`%s': Received `%s' request from client, key %s (msg size %d, we expected %d)\n", "DHT", "FIND PEER", GNUNET_h2s (message_context->key), ntohs (find_msg->header.size), sizeof (struct GNUNET_DHT_FindPeerMessage)); #endif GNUNET_assert (ntohs (find_msg->header.size) >= sizeof (struct GNUNET_DHT_FindPeerMessage)); /* FIXME: Implement find peer functionality here */ } /** * Server handler for initiating local dht put requests */ static void handle_dht_put (void *cls, struct GNUNET_DHT_PutMessage *put_msg, struct DHT_MessageContext *message_context) { size_t put_type; size_t data_size; GNUNET_assert (ntohs (put_msg->header.size) >= sizeof (struct GNUNET_DHT_PutMessage)); put_type = ntohs (put_msg->type); data_size = ntohs (put_msg->data_size); #if DEBUG_DHT GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "`%s': %s msg total size is %d, data size %d, struct size %d\n", "DHT", "PUT", ntohs (put_msg->header.size), data_size, sizeof (struct GNUNET_DHT_PutMessage)); #endif GNUNET_assert (ntohs (put_msg->header.size) == sizeof (struct GNUNET_DHT_PutMessage) + data_size); #if DEBUG_DHT GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "`%s': Received `%s' request from client, message type %d, key %s\n", "DHT", "PUT", put_type, GNUNET_h2s (message_context->key)); #endif /** * Simplest DHT functionality, store any message we receive a put request for. */ if (datacache != NULL) GNUNET_DATACACHE_put (datacache, message_context->key, data_size, (char *) &put_msg[1], put_type, put_msg->expiration); /** * FIXME: Implement dht put request functionality here! */ } /** * Context for sending receipt confirmations. Not used yet. */ struct SendConfirmationContext { /** * The message to send. */ struct GNUNET_DHT_StopMessage *message; /** * Transmit handle. */ struct GNUNET_CONNECTION_TransmitHandle *transmit_handle; }; static size_t send_confirmation (void *cls, size_t size, void *buf) { struct GNUNET_DHT_StopMessage *confirmation_message = cls; if (buf == NULL) /* Message timed out, that's crappy... */ { GNUNET_free (confirmation_message); return 0; } if (size >= ntohs (confirmation_message->header.size)) { memcpy (buf, confirmation_message, ntohs (confirmation_message->header.size)); return ntohs (confirmation_message->header.size); } else return 0; } static void send_client_receipt_confirmation (struct GNUNET_SERVER_Client *client, uint64_t uid) { struct GNUNET_DHT_StopMessage *confirm_message; #if DEBUG_DHT GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "`%s': Sending receipt confirmation for uid %llu\n", "DHT", uid); #endif confirm_message = GNUNET_malloc (sizeof (struct GNUNET_DHT_StopMessage)); confirm_message->header.type = htons (GNUNET_MESSAGE_TYPE_DHT_STOP); confirm_message->header.size = htons (sizeof (struct GNUNET_DHT_StopMessage)); confirm_message->unique_id = GNUNET_htonll (uid); GNUNET_SERVER_notify_transmit_ready (client, sizeof (struct GNUNET_DHT_StopMessage), GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 5), &send_confirmation, confirm_message); } static void handle_dht_start_message (void *cls, struct GNUNET_SERVER_Client *client, const struct GNUNET_MessageHeader *message) { struct GNUNET_DHT_Message *dht_msg = (struct GNUNET_DHT_Message *) message; struct GNUNET_MessageHeader *enc_msg; struct DHT_MessageContext *message_context; size_t enc_type; enc_msg = (struct GNUNET_MessageHeader *) &dht_msg[1]; enc_type = ntohs (enc_msg->type); #if DEBUG_DHT GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "`%s': Received `%s' request from client, message type %d, key %s, uid %llu\n", "DHT", "GENERIC", enc_type, GNUNET_h2s (&dht_msg->key), GNUNET_ntohll (dht_msg->unique_id)); #endif message_context = GNUNET_malloc (sizeof (struct DHT_MessageContext)); message_context->client = client; message_context->key = &dht_msg->key; message_context->unique_id = GNUNET_ntohll (dht_msg->unique_id); message_context->replication = ntohs (dht_msg->desired_replication_level); message_context->msg_options = ntohs (dht_msg->options); switch (enc_type) { case GNUNET_MESSAGE_TYPE_DHT_GET: handle_dht_get (cls, (struct GNUNET_DHT_GetMessage *) enc_msg, message_context); break; case GNUNET_MESSAGE_TYPE_DHT_PUT: handle_dht_put (cls, (struct GNUNET_DHT_PutMessage *) enc_msg, message_context); send_client_receipt_confirmation (client, GNUNET_ntohll (dht_msg->unique_id)); break; case GNUNET_MESSAGE_TYPE_DHT_FIND_PEER: handle_dht_find_peer (cls, (struct GNUNET_DHT_FindPeerMessage *) enc_msg, message_context); break; default: GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "`%s': Message type (%d) not handled\n", "DHT", enc_type); } GNUNET_free (message_context); GNUNET_SERVER_receive_done (client, GNUNET_OK); } static void handle_dht_stop_message (void *cls, struct GNUNET_SERVER_Client *client, const struct GNUNET_MessageHeader *message) { struct GNUNET_DHT_StopMessage *dht_stop_msg = (struct GNUNET_DHT_StopMessage *) message; #if DEBUG_DHT GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "`%s': Received `%s' request from client, uid %llu\n", "DHT", "GENERIC STOP", GNUNET_ntohll (dht_stop_msg->unique_id)); #endif /* TODO: Put in demultiplexing here */ send_client_receipt_confirmation (client, GNUNET_ntohll (dht_stop_msg->unique_id)); GNUNET_SERVER_receive_done (client, GNUNET_OK); } /** * Core handler for p2p dht get requests. */ static int handle_dht_p2p_get (void *cls, const struct GNUNET_PeerIdentity *peer, const struct GNUNET_MessageHeader *message, struct GNUNET_TIME_Relative latency, uint32_t distance) { #if DEBUG_DHT GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "`%s': Received `%s' request from another peer\n", "DHT", "GET"); #endif return GNUNET_YES; } /** * Core handler for p2p dht put requests. */ static int handle_dht_p2p_put (void *cls, const struct GNUNET_PeerIdentity *peer, const struct GNUNET_MessageHeader *message, struct GNUNET_TIME_Relative latency, uint32_t distance) { #if DEBUG_DHT GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "`%s': Received `%s' request from another peer\n", "DHT", "PUT"); #endif return GNUNET_YES; } /** * Core handler for p2p dht find peer requests. */ static int handle_dht_p2p_find_peer (void *cls, const struct GNUNET_PeerIdentity *peer, const struct GNUNET_MessageHeader *message, struct GNUNET_TIME_Relative latency, uint32_t distance) { #if DEBUG_DHT GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "`%s': Received `%s' request from another peer\n", "DHT", "FIND PEER"); #endif return GNUNET_YES; } /** * Task run during shutdown. * * @param cls unused * @param tc unused */ static void shutdown_task (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc) { GNUNET_CORE_disconnect (coreAPI); } /** * To be called on core init/fail. */ void core_init (void *cls, struct GNUNET_CORE_Handle *server, const struct GNUNET_PeerIdentity *identity, const struct GNUNET_CRYPTO_RsaPublicKeyBinaryEncoded *publicKey) { if (server == NULL) { GNUNET_SCHEDULER_cancel (sched, cleanup_task); GNUNET_SCHEDULER_add_now (sched, &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 memcpy (&my_identity, identity, sizeof (struct GNUNET_PeerIdentity)); coreAPI = server; } /** * Process dht requests. * * @param cls closure * @param scheduler scheduler to use * @param server the initialized server * @param c configuration to use */ static void run (void *cls, struct GNUNET_SCHEDULER_Handle *scheduler, struct GNUNET_SERVER_Handle *server, const struct GNUNET_CONFIGURATION_Handle *c) { sched = scheduler; cfg = c; datacache = GNUNET_DATACACHE_create (sched, cfg, "dhtcache"); client_transmit_timeout = GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 5); GNUNET_SERVER_add_handlers (server, plugin_handlers); coreAPI = GNUNET_CORE_connect (sched, /* Main scheduler */ cfg, /* Main configuration */ client_transmit_timeout, /* Delay for connecting */ NULL, /* FIXME: anything we want to pass around? */ &core_init, /* Call core_init once connected */ NULL, /* Don't care about pre-connects */ NULL, /* Don't care about connects */ NULL, /* Don't care about disconnects */ 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; /* Scheduled the task to clean up when shutdown is called */ cleanup_task = GNUNET_SCHEDULER_add_delayed (sched, 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) { return (GNUNET_OK == GNUNET_SERVICE_run (argc, argv, "dht", GNUNET_SERVICE_OPTION_NONE, &run, NULL)) ? 0 : 1; }