X-Git-Url: https://git.librecmc.org/?a=blobdiff_plain;f=src%2Fdv%2Fgnunet-service-dv.c;h=844e44b6ffb4573962a6cdb5b9920bb3655c669b;hb=19377520016cc644070d207af34ae3e76618fdc8;hp=72dc402888b84da29b44192400b04ddc97497afb;hpb=f7e6a15e96565fa49ed62402b503254981c2b464;p=oweals%2Fgnunet.git diff --git a/src/dv/gnunet-service-dv.c b/src/dv/gnunet-service-dv.c index 72dc40288..844e44b6f 100644 --- a/src/dv/gnunet-service-dv.c +++ b/src/dv/gnunet-service-dv.c @@ -4,7 +4,7 @@ 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 + 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 @@ -27,9 +27,6 @@ * @author Christian Grothoff * @author Nathan Evans * - * TODO: The gossip rates need to be worked out. Probably many other things - * as well. - * */ #include "platform.h" #include "gnunet_client_lib.h" @@ -43,39 +40,21 @@ #include "gnunet_hello_lib.h" #include "gnunet_peerinfo_service.h" #include "gnunet_crypto_lib.h" +#include "gnunet_statistics_service.h" #include "dv.h" /** * For testing mostly, remember only the * shortest path to a distant neighbor. */ -#define AT_MOST_ONE GNUNET_YES +#define AT_MOST_ONE GNUNET_NO #define USE_PEER_ID GNUNET_YES /** - * DV Service Context stuff goes here... - */ - -/** - * Handle to the core service api. - */ -static struct GNUNET_CORE_Handle *coreAPI; - -/** - * The identity of our peer. - */ -static struct GNUNET_PeerIdentity my_identity; - -/** - * The configuration for this service. - */ -static const struct GNUNET_CONFIGURATION_Handle *cfg; - -/** - * The scheduler for this service. + * How many outstanding messages (unknown sender) will we allow per peer? */ -static struct GNUNET_SCHEDULER_Handle *sched; +#define MAX_OUTSTANDING_MESSAGES 5 /** * How often do we check about sending out more peer information (if @@ -136,33 +115,6 @@ static struct GNUNET_SCHEDULER_Handle *sched; */ #define DEFAULT_FISHEYE_DEPTH 4 -/** - * The client, the DV plugin connected to us. Hopefully - * this client will never change, although if the plugin dies - * and returns for some reason it may happen. - */ -static struct GNUNET_SERVER_Client * client_handle; - -/** - * Task to run when we shut down, cleaning up all our trash - */ -GNUNET_SCHEDULER_TaskIdentifier cleanup_task; - -/** - * Task to run to gossip about peers. Will reschedule itself forever until shutdown! - */ -GNUNET_SCHEDULER_TaskIdentifier gossip_task; - -/** - * Struct where neighbor information is stored. - */ -struct DistantNeighbor *referees; - -static size_t default_dv_priority = 0; - -char *my_short_id; - - /** * Linked list of messages to send to clients. */ @@ -210,42 +162,6 @@ struct PendingMessage }; -/** - * Transmit handle to the plugin. - */ -struct GNUNET_CONNECTION_TransmitHandle * plugin_transmit_handle; - -/** - * Head of DLL for client messages - */ -struct PendingMessage *plugin_pending_head; - -/** - * Tail of DLL for client messages - */ -struct PendingMessage *plugin_pending_tail; - -/** - * Handle to the peerinfo service - */ -struct GNUNET_PEERINFO_Handle *peerinfo_handle; - -/** - * Transmit handle to core service. - */ -struct GNUNET_CORE_TransmitHandle * core_transmit_handle; - -/** - * Head of DLL for core messages - */ -struct PendingMessage *core_pending_head; - -/** - * Tail of DLL for core messages - */ -struct PendingMessage *core_pending_tail; - - struct FastGossipNeighborList { /** @@ -329,6 +245,38 @@ struct NeighborUpdateInfo }; +/** + * Struct to store a single message received with + * an unknown sender. + */ +struct UnknownSenderMessage +{ + /** + * Message sender (immediate) + */ + struct GNUNET_PeerIdentity sender; + + /** + * The actual message received + */ + struct GNUNET_MessageHeader *message; + + /** + * Latency of connection + */ + struct GNUNET_TIME_Relative latency; + + /** + * Distance to destination + */ + uint32_t distance; + + /** + * Unknown sender id + */ + uint32_t sender_id; +}; + /** * Struct where actual neighbor information is stored, * referenced by min_heap and max_heap. Freeing dealt @@ -366,6 +314,13 @@ struct DirectNeighbor * from DV? */ int hidden; + + /** + * Save messages immediately from this direct neighbor from a + * distan peer we don't know on the chance that it will be + * gossiped about and we can deliver the message. + */ + struct UnknownSenderMessage pending_messages[MAX_OUTSTANDING_MESSAGES]; }; @@ -416,6 +371,11 @@ struct DistantNeighbor */ struct GNUNET_TIME_Absolute last_activity; + /** + * Last time we sent routing information about this peer + */ + struct GNUNET_TIME_Absolute last_gossip; + /** * Cost to neighbor, used for actual distance vector computations */ @@ -522,58 +482,12 @@ struct DV_SendContext */ struct GNUNET_TIME_Relative timeout; -#if DEBUG_DV_MESSAGES /** * Unique ID for DV message */ unsigned int uid; -#endif -}; - -/** - * Global construct - */ -struct GNUNET_DV_Context -{ - /** - * Map of PeerIdentifiers to 'struct GNUNET_dv_neighbor*'s for all - * directly connected peers. - */ - struct GNUNET_CONTAINER_MultiHashMap *direct_neighbors; - - /** - * Map of PeerIdentifiers to 'struct GNUNET_dv_neighbor*'s for - * peers connected via DV (extended neighborhood). Does ALSO - * include any peers that are in 'direct_neighbors'; for those - * peers, the cost will be zero and the referrer all zeros. - */ - struct GNUNET_CONTAINER_MultiHashMap *extended_neighbors; - - /** - * We use the min heap (min refers to cost) to prefer - * gossipping about peers with small costs. - */ - struct GNUNET_CONTAINER_Heap *neighbor_min_heap; - - /** - * We use the max heap (max refers to cost) for general - * iterations over all peers and to remove the most costly - * connection if we have too many. - */ - struct GNUNET_CONTAINER_Heap *neighbor_max_heap; - - unsigned long long fisheye_depth; - - unsigned long long max_table_size; - - unsigned int neighbor_id_loc; - - int closing; - }; -static struct GNUNET_DV_Context ctx; - struct FindDestinationContext { unsigned int tid; @@ -600,6 +514,148 @@ struct DisconnectContext struct DirectNeighbor *direct; }; +struct TokenizedMessageContext +{ + /** + * Immediate sender of this message + */ + const struct GNUNET_PeerIdentity *peer; + + /** + * Distant sender of the message + */ + struct DistantNeighbor *distant; + + /** + * Uid for this set of messages + */ + uint32_t uid; +}; + +/** + * Context for finding the least cost peer to send to. + * Transport selection can only go so far. + */ +struct FindLeastCostContext +{ + struct DistantNeighbor *target; + unsigned int least_cost; +}; + +/** + * Handle to the core service api. + */ +static struct GNUNET_CORE_Handle *coreAPI; + +/** + * Stream tokenizer to handle messages coming in from core. + */ +static struct GNUNET_SERVER_MessageStreamTokenizer *coreMST; + +/** + * The identity of our peer. + */ +static struct GNUNET_PeerIdentity my_identity; + +/** + * The configuration for this service. + */ +static const struct GNUNET_CONFIGURATION_Handle *cfg; + + +/** + * The client, the DV plugin connected to us. Hopefully + * this client will never change, although if the plugin dies + * and returns for some reason it may happen. + */ +static struct GNUNET_SERVER_Client * client_handle; + +/** + * Task to run when we shut down, cleaning up all our trash + */ +static GNUNET_SCHEDULER_TaskIdentifier cleanup_task; + +static size_t default_dv_priority = 0; + +static char *my_short_id; + +/** + * Transmit handle to the plugin. + */ +static struct GNUNET_CONNECTION_TransmitHandle * plugin_transmit_handle; + +/** + * Head of DLL for client messages + */ +static struct PendingMessage *plugin_pending_head; + +/** + * Tail of DLL for client messages + */ +static struct PendingMessage *plugin_pending_tail; + +/** + * Handle to the peerinfo service + */ +static struct GNUNET_PEERINFO_Handle *peerinfo_handle; + +/** + * Transmit handle to core service. + */ +static struct GNUNET_CORE_TransmitHandle * core_transmit_handle; + +/** + * Head of DLL for core messages + */ +static struct PendingMessage *core_pending_head; + +/** + * Tail of DLL for core messages + */ +static struct PendingMessage *core_pending_tail; + +/** + * Map of PeerIdentifiers to 'struct GNUNET_dv_neighbor*'s for all + * directly connected peers. + */ +static struct GNUNET_CONTAINER_MultiHashMap *direct_neighbors; + +/** + * Map of PeerIdentifiers to 'struct GNUNET_dv_neighbor*'s for + * peers connected via DV (extended neighborhood). Does ALSO + * include any peers that are in 'direct_neighbors'; for those + * peers, the cost will be zero and the referrer all zeros. + */ +static struct GNUNET_CONTAINER_MultiHashMap *extended_neighbors; + +/** + * We use the min heap (min refers to cost) to prefer + * gossipping about peers with small costs. + */ +static struct GNUNET_CONTAINER_Heap *neighbor_min_heap; + +/** + * We use the max heap (max refers to cost) for general + * iterations over all peers and to remove the most costly + * connection if we have too many. + */ +static struct GNUNET_CONTAINER_Heap *neighbor_max_heap; + +/** + * Handle for the statistics service. + */ +struct GNUNET_STATISTICS_Handle *stats; + +/** + * How far out to keep peers we learn about. + */ +static unsigned long long fisheye_depth; + +/** + * How many peers to store at most. + */ +static unsigned long long max_table_size; + /** * We've been given a target ID based on the random numbers that * we assigned to our DV-neighborhood. Find the entry for the @@ -685,8 +741,8 @@ size_t transmit_to_plugin (void *cls, if (buf == NULL) { /* client disconnected */ -#if DEBUG_DV - GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "`%s': buffer was NULL\n", "DHT"); +#if DEBUG_DV_MESSAGES + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "%s: %s buffer was NULL (client disconnect?)\n", my_short_id, "transmit_to_plugin"); #endif return 0; } @@ -695,9 +751,6 @@ size_t transmit_to_plugin (void *cls, while ( (NULL != (reply = plugin_pending_head)) && (size >= off + (msize = ntohs (reply->msg->size)))) { -#if DEBUG_DV - GNUNET_log(GNUNET_ERROR_TYPE_DEBUG, "`%s' : transmit_notify (plugin) called with size %d\n", "dv service", msize); -#endif GNUNET_CONTAINER_DLL_remove (plugin_pending_head, plugin_pending_tail, reply); @@ -733,11 +786,6 @@ void send_to_plugin(const struct GNUNET_PeerIdentity * sender, { struct GNUNET_DV_MessageReceived *received_msg; struct PendingMessage *pending_message; -#if DEBUG_DV - struct GNUNET_MessageHeader * packed_message_header; - struct GNUNET_HELLO_Message *hello_msg; - struct GNUNET_PeerIdentity hello_identity; -#endif char *sender_address; size_t sender_address_len; char *packed_msg_start; @@ -765,7 +813,6 @@ void send_to_plugin(const struct GNUNET_PeerIdentity * sender, received_msg = GNUNET_malloc(size); received_msg->header.size = htons(size); received_msg->header.type = htons(GNUNET_MESSAGE_TYPE_TRANSPORT_DV_RECEIVE); - received_msg->sender_address_len = htonl(sender_address_len); received_msg->distance = htonl(cost); received_msg->msg_len = htonl(message_size); /* Set the sender in this message to be the original sender! */ @@ -777,17 +824,6 @@ void send_to_plugin(const struct GNUNET_PeerIdentity * sender, packed_msg_start = (char *)&received_msg[1]; packed_msg_start = &packed_msg_start[sender_address_len]; memcpy(packed_msg_start, message, message_size); -#if DEBUG_DV - packed_message_header = (struct GNUNET_MessageHeader *)packed_msg_start; - GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "dv service created received message. sender_address_len %lu, packed message len %d, total len %d\n", sender_address_len, ntohl(received_msg->msg_len), ntohs(received_msg->header.size)); - GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "dv packed message len %d, type %d\n", ntohs(packed_message_header->size), ntohs(packed_message_header->type)); - if (ntohs(packed_message_header->type) == GNUNET_MESSAGE_TYPE_HELLO) - { - hello_msg = (struct GNUNET_HELLO_Message *)packed_message_header; - GNUNET_assert(GNUNET_OK == GNUNET_HELLO_get_id(hello_msg, &hello_identity)); - GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Packed HELLO message is about peer %s\n", GNUNET_i2s(&hello_identity)); - } -#endif pending_message = GNUNET_malloc(sizeof(struct PendingMessage) + size); pending_message->msg = (struct GNUNET_MessageHeader *)&pending_message[1]; memcpy(&pending_message[1], received_msg, size); @@ -803,15 +839,40 @@ void send_to_plugin(const struct GNUNET_PeerIdentity * sender, size, GNUNET_TIME_UNIT_FOREVER_REL, &transmit_to_plugin, NULL); } -#if DEBUG_DV - else - { - GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Failed to queue message for plugin, must be one in progress already!!\n"); - } -#endif + } + else + { + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Failed to queue message for plugin, client_handle not yet set (how?)!\n"); } } +/* Declare here so retry_core_send is aware of it */ +size_t core_transmit_notify (void *cls, + size_t size, void *buf); + +/** + * Try to send another message from our core sending list + */ +static void +try_core_send (void *cls, + const struct GNUNET_SCHEDULER_TaskContext *tc) +{ + struct PendingMessage *pending; + pending = core_pending_head; + + if (core_transmit_handle != NULL) + return; /* Message send already in progress */ + + if ((pending != NULL) && (coreAPI != NULL)) + core_transmit_handle = GNUNET_CORE_notify_transmit_ready (coreAPI, + GNUNET_YES, + pending->importance, + pending->timeout, + &pending->recipient, + pending->msg_size, + &core_transmit_notify, NULL); +} + /** * Function called to notify a client about the socket @@ -819,7 +880,7 @@ void send_to_plugin(const struct GNUNET_PeerIdentity * sender, * NULL and "size" zero if the socket was closed for * writing in the meantime. * - * @param cls closure + * @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 @@ -828,7 +889,7 @@ size_t core_transmit_notify (void *cls, size_t size, void *buf) { char *cbuf = buf; - struct PendingMessage *reply; + struct PendingMessage *pending; struct PendingMessage *client_reply; size_t off; size_t msize; @@ -844,22 +905,22 @@ size_t core_transmit_notify (void *cls, core_transmit_handle = NULL; off = 0; - reply = core_pending_head; - if ( (reply != NULL) && - (size >= (msize = ntohs (reply->msg->size)))) + pending = core_pending_head; + if ( (pending != NULL) && + (size >= (msize = ntohs (pending->msg->size)))) { #if DEBUG_DV GNUNET_log(GNUNET_ERROR_TYPE_DEBUG, "`%s' : transmit_notify (core) called with size %d\n", "dv service", msize); #endif GNUNET_CONTAINER_DLL_remove (core_pending_head, core_pending_tail, - reply); - if (reply->send_result != NULL) /* Will only be non-null if a real client asked for this send */ + pending); + if (pending->send_result != NULL) /* Will only be non-null if a real client asked for this send */ { client_reply = GNUNET_malloc(sizeof(struct PendingMessage) + sizeof(struct GNUNET_DV_SendResultMessage)); client_reply->msg = (struct GNUNET_MessageHeader *)&client_reply[1]; - memcpy(&client_reply[1], reply->send_result, sizeof(struct GNUNET_DV_SendResultMessage)); - GNUNET_free(reply->send_result); + memcpy(&client_reply[1], pending->send_result, sizeof(struct GNUNET_DV_SendResultMessage)); + GNUNET_free(pending->send_result); GNUNET_CONTAINER_DLL_insert_after(plugin_pending_head, plugin_pending_tail, plugin_pending_tail, client_reply); if (client_handle != NULL) @@ -877,13 +938,15 @@ size_t core_transmit_notify (void *cls, } } } - memcpy (&cbuf[off], reply->msg, msize); - GNUNET_free (reply); + memcpy (&cbuf[off], pending->msg, msize); + GNUNET_free (pending); off += msize; } - reply = core_pending_head; - if (reply != NULL) - core_transmit_handle = GNUNET_CORE_notify_transmit_ready(coreAPI, reply->importance, reply->timeout, &reply->recipient, reply->msg_size, &core_transmit_notify, NULL); + /*reply = core_pending_head;*/ + + GNUNET_SCHEDULER_add_now(&try_core_send, NULL); + /*if (reply != NULL) + core_transmit_handle = GNUNET_CORE_notify_transmit_ready(coreAPI, GNUNET_YES, reply->importance, reply->timeout, &reply->recipient, reply->msg_size, &core_transmit_notify, NULL);*/ return off; } @@ -917,10 +980,7 @@ send_message_via (const struct GNUNET_PeerIdentity *sender, find_context.dest = send_context->distant_peer; find_context.via = recipient; find_context.tid = 0; - //specific_neighbor = GNUNET_CONTAINER_multihashmap_get(ctx.extended_neighbors, &send_context->distant_peer->hashPubKey); - - //GNUNET_CONTAINER_multihashmap_iterate(ctx.extended_neighbors, &find_specific_id, &find_context); - GNUNET_CONTAINER_multihashmap_get_multiple (ctx.extended_neighbors, &send_context->distant_peer->hashPubKey, + GNUNET_CONTAINER_multihashmap_get_multiple (extended_neighbors, &send_context->distant_peer->hashPubKey, &find_specific_id, &find_context); if (find_context.tid == 0) @@ -935,14 +995,14 @@ send_message_via (const struct GNUNET_PeerIdentity *sender, sender, sizeof (struct GNUNET_PeerIdentity)))) { sender_id = 0; - source = GNUNET_CONTAINER_multihashmap_get (ctx.extended_neighbors, + source = GNUNET_CONTAINER_multihashmap_get (extended_neighbors, &sender->hashPubKey); if (source != NULL) GNUNET_log(GNUNET_ERROR_TYPE_WARNING, "%s: send_message_via found %s, myself in extended peer list???\n", my_short_id, GNUNET_i2s(&source->identity)); } else { - source = GNUNET_CONTAINER_multihashmap_get (ctx.extended_neighbors, + source = GNUNET_CONTAINER_multihashmap_get (extended_neighbors, &sender->hashPubKey); if (source == NULL) { @@ -966,6 +1026,8 @@ send_message_via (const struct GNUNET_PeerIdentity *sender, toSend->recipient = htonl (recipient_id); #if DEBUG_DV_MESSAGES toSend->uid = send_context->uid; /* Still sent around in network byte order */ +#else + toSend->uid = htonl(0); #endif memcpy (&toSend[1], send_context->message, send_context->message_size); @@ -981,29 +1043,11 @@ send_message_via (const struct GNUNET_PeerIdentity *sender, core_pending_tail, pending_message); - if (core_transmit_handle == NULL) - core_transmit_handle = GNUNET_CORE_notify_transmit_ready(coreAPI, send_context->importance, send_context->timeout, recipient, msg_size, &core_transmit_notify, NULL); - else - { -#if DEBUG_DV - GNUNET_log(GNUNET_ERROR_TYPE_DEBUG, "`%s': Failed to schedule pending transmission (must be one in progress!)\n", "dv service"); -#endif - } + GNUNET_SCHEDULER_add_now(try_core_send, NULL); + return GNUNET_YES; } - -/** - * Context for finding the least cost peer to send to. - * Transport selection can only go so far. - */ -struct FindLeastCostContext -{ - struct DistantNeighbor *target; - unsigned int least_cost; -}; - - /** * Given a FindLeastCostContext, and a set * of peers that match the target, return the cheapest. @@ -1025,26 +1069,12 @@ find_least_cost_peer (void *cls, if (dn->cost < find_context->least_cost) { find_context->target = dn; - } - if (dn->cost == DIRECT_NEIGHBOR_COST) - return GNUNET_NO; - return GNUNET_YES; -} - -#if DEBUG_DV_MESSAGES -/** - * Send a DV data message via DV. - * - * @param recipient the ultimate recipient of this message - * @param sender the original sender of the message - * @param specific_neighbor the specific neighbor to send this message via - * @param message the packed message - * @param message_size size of the message - * @param importance what priority to send this message with - * @param uid unique id for this message - * @param timeout how long to possibly delay sending this message - */ -#else + } + if (dn->cost == DIRECT_NEIGHBOR_COST) + return GNUNET_NO; + return GNUNET_YES; +} + /** * Send a DV data message via DV. * @@ -1054,9 +1084,9 @@ find_least_cost_peer (void *cls, * @param message the packed message * @param message_size size of the message * @param importance what priority to send this message with + * @param uid the unique identifier of this message (or 0 for none) * @param timeout how long to possibly delay sending this message */ -#endif static int send_message (const struct GNUNET_PeerIdentity * recipient, const struct GNUNET_PeerIdentity * sender, @@ -1064,9 +1094,7 @@ send_message (const struct GNUNET_PeerIdentity * recipient, const struct GNUNET_MessageHeader * message, size_t message_size, unsigned int importance, -#if DEBUG_DV_MESSAGES unsigned int uid, -#endif struct GNUNET_TIME_Relative timeout) { p2p_dv_MESSAGE_Data *toSend; @@ -1093,7 +1121,7 @@ send_message (const struct GNUNET_PeerIdentity * recipient, * in messages looping forever. Relatively cheap, we don't iterate * over all known peers, just those that apply. */ - GNUNET_CONTAINER_multihashmap_get_multiple (ctx.extended_neighbors, + GNUNET_CONTAINER_multihashmap_get_multiple (extended_neighbors, &recipient->hashPubKey, &find_least_cost_peer, &find_least_ctx); target = find_least_ctx.target; @@ -1104,8 +1132,8 @@ send_message (const struct GNUNET_PeerIdentity * recipient, } recipient_id = target->referrer_id; - source = GNUNET_CONTAINER_multihashmap_get (ctx.extended_neighbors, - &sender->hashPubKey); + source = GNUNET_CONTAINER_multihashmap_get (extended_neighbors, + &sender->hashPubKey); if (source == NULL) { if (0 != (memcmp (&my_identity, @@ -1129,7 +1157,7 @@ send_message (const struct GNUNET_PeerIdentity * recipient, encPeerFrom.encoding[4] = '\0'; encPeerVia.encoding[4] = '\0'; #endif - if (0 == memcmp(&source->identity, &target->referrer->identity, sizeof(struct GNUNET_PeerIdentity))) + if ((sender_id != 0) && (0 == memcmp(&source->identity, &target->referrer->identity, sizeof(struct GNUNET_PeerIdentity)))) { return 0; } @@ -1149,13 +1177,20 @@ send_message (const struct GNUNET_PeerIdentity * recipient, toSend->recipient = htonl (recipient_id); #if DEBUG_DV_MESSAGES toSend->uid = htonl(uid); +#else + toSend->uid = htonl(0); #endif + #if DEBUG_DV_PEER_NUMBERS GNUNET_CRYPTO_hash_to_enc (&target->identity.hashPubKey, &encPeerTo); encPeerTo.encoding[4] = '\0'; GNUNET_log(GNUNET_ERROR_TYPE_DEBUG, "%s: Sending DATA message. Sender id %u, source %s, destination %s, via %s\n", GNUNET_i2s(&my_identity), sender_id, &encPeerFrom, &encPeerTo, &encPeerVia); #endif memcpy (&toSend[1], message, message_size); + if ((source != NULL) && (source->pkey == NULL)) /* Test our hypothesis about message failures! */ + { + GNUNET_log(GNUNET_ERROR_TYPE_WARNING, "%s: Sending message, but anticipate recipient will not know sender!!!\n\n\n", my_short_id); + } GNUNET_CONTAINER_DLL_insert_after (core_pending_head, core_pending_tail, core_pending_tail, @@ -1163,10 +1198,8 @@ send_message (const struct GNUNET_PeerIdentity * recipient, #if DEBUG_DV GNUNET_log(GNUNET_ERROR_TYPE_DEBUG, "%s: Notifying core of send size %d to destination `%s'\n", "DV SEND MESSAGE", msg_size, GNUNET_i2s(recipient)); #endif - if (core_transmit_handle == NULL) - core_transmit_handle = GNUNET_CORE_notify_transmit_ready(coreAPI, importance, timeout, &target->referrer->identity, msg_size, &core_transmit_notify, NULL); - else - GNUNET_log(GNUNET_ERROR_TYPE_DEBUG, "%s: CORE ALREADY SENDING\n", "DV SEND MESSAGE", msg_size); + + GNUNET_SCHEDULER_add_now(try_core_send, NULL); return (int) cost; } @@ -1211,6 +1244,107 @@ int checkPeerID (void *cls, } #endif + +/** + * Handler for messages parsed out by the tokenizer from + * DV DATA received for this peer. + * + * @param cls NULL + * @param client the TokenizedMessageContext which contains message information + * @param message the actual message + */ +void tokenized_message_handler (void *cls, + void *client, + const struct GNUNET_MessageHeader *message) +{ + struct TokenizedMessageContext *ctx = client; + GNUNET_break_op (ntohs (message->type) != GNUNET_MESSAGE_TYPE_DV_GOSSIP); + GNUNET_break_op (ntohs (message->type) != GNUNET_MESSAGE_TYPE_DV_DATA); + if ( (ntohs (message->type) != GNUNET_MESSAGE_TYPE_DV_GOSSIP) && + (ntohs (message->type) != GNUNET_MESSAGE_TYPE_DV_DATA) ) + { +#if DEBUG_DV_MESSAGES + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, + "%s: Receives %s message for me, uid %u, size %d, type %d cost %u from %s!\n", my_short_id, "DV DATA", ctx->uid, ntohs(message->size), ntohs(message->type), ctx->distant->cost, GNUNET_i2s(&ctx->distant->identity)); +#endif + GNUNET_assert(memcmp(ctx->peer, &ctx->distant->identity, sizeof(struct GNUNET_PeerIdentity)) != 0); + send_to_plugin(ctx->peer, message, ntohs(message->size), &ctx->distant->identity, ctx->distant->cost); + } +} + +#if DELAY_FORWARDS +struct DelayedMessageContext +{ + struct GNUNET_PeerIdentity dest; + struct GNUNET_PeerIdentity sender; + struct GNUNET_MessageHeader *message; + size_t message_size; + uint32_t uid; +}; + +void send_message_delayed (void *cls, + const struct GNUNET_SCHEDULER_TaskContext *tc) +{ + struct DelayedMessageContext *msg_ctx = cls; + if (msg_ctx != NULL) + { + send_message(&msg_ctx->dest, + &msg_ctx->sender, + NULL, + msg_ctx->message, + msg_ctx->message_size, + default_dv_priority, + msg_ctx->uid, + GNUNET_TIME_relative_get_forever()); + GNUNET_free(msg_ctx->message); + GNUNET_free(msg_ctx); + } +} +#endif + +/** + * Get distance information from 'atsi'. + * + * @param atsi performance data + * @return connected transport distance + */ +static uint32_t +get_atsi_distance (const struct GNUNET_TRANSPORT_ATS_Information *atsi) +{ + while ( (ntohl (atsi->type) != GNUNET_TRANSPORT_ATS_ARRAY_TERMINATOR) && + (ntohl (atsi->type) != GNUNET_TRANSPORT_ATS_QUALITY_NET_DISTANCE) ) + atsi++; + if (ntohl (atsi->type) == GNUNET_TRANSPORT_ATS_ARRAY_TERMINATOR) + { + GNUNET_break (0); + /* FIXME: we do not have distance data? Assume direct neighbor. */ + return DIRECT_NEIGHBOR_COST; + } + return ntohl (atsi->value); +} + +/** + * Find latency information in 'atsi'. + * + * @param atsi performance data + * @return connection latency + */ +static struct GNUNET_TIME_Relative +get_atsi_latency (const struct GNUNET_TRANSPORT_ATS_Information *atsi) +{ + while ( (ntohl (atsi->type) != GNUNET_TRANSPORT_ATS_ARRAY_TERMINATOR) && + (ntohl (atsi->type) != GNUNET_TRANSPORT_ATS_QUALITY_NET_DELAY) ) + atsi++; + if (ntohl (atsi->type) == GNUNET_TRANSPORT_ATS_ARRAY_TERMINATOR) + { + GNUNET_break (0); + /* how can we not have latency data? */ + return GNUNET_TIME_UNIT_SECONDS; + } + /* FIXME: Multiply by GNUNET_TIME_UNIT_MILLISECONDS (1) to get as a GNUNET_TIME_Relative */ + return GNUNET_TIME_relative_multiply(GNUNET_TIME_UNIT_MILLISECONDS, ntohl (atsi->value)); +} + /** * Core handler for dv data messages. Whatever this message * contains all we really have to do is rip it out of its @@ -1220,14 +1354,13 @@ int checkPeerID (void *cls, * @param cls closure * @param peer peer which sent the message (immediate sender) * @param message the message - * @param latency the latency of the connection we received the message from - * @param distance the distance to the immediate peer - */ -static int handle_dv_data_message (void *cls, - const struct GNUNET_PeerIdentity * peer, - const struct GNUNET_MessageHeader * message, - struct GNUNET_TIME_Relative latency, - uint32_t distance) + * @param atsi transport ATS information (latency, distance, etc.) + */ +static int +handle_dv_data_message (void *cls, + const struct GNUNET_PeerIdentity * peer, + const struct GNUNET_MessageHeader * message, + const struct GNUNET_TRANSPORT_ATS_Information *atsi) { const p2p_dv_MESSAGE_Data *incoming = (const p2p_dv_MESSAGE_Data *) message; const struct GNUNET_MessageHeader *packed_message; @@ -1235,22 +1368,28 @@ static int handle_dv_data_message (void *cls, struct DistantNeighbor *pos; unsigned int sid; /* Sender id */ unsigned int tid; /* Target id */ - struct GNUNET_PeerIdentity original_sender; - struct GNUNET_PeerIdentity destination; + struct GNUNET_PeerIdentity *original_sender; + struct GNUNET_PeerIdentity *destination; struct FindDestinationContext fdc; + struct TokenizedMessageContext tkm_ctx; + int i; + int found_pos; +#if DELAY_FORWARDS + struct DelayedMessageContext *delayed_context; +#endif #if USE_PEER_ID struct CheckPeerContext checkPeerCtx; #endif +#if DEBUG_DV_MESSAGES char *sender_id; - char *direct_id; +#endif int ret; size_t packed_message_size; char *cbuf; - size_t offset; + uint32_t distance; /* Distance information */ + struct GNUNET_TIME_Relative latency; /* Latency information */ packed_message_size = ntohs(incoming->header.size) - sizeof(p2p_dv_MESSAGE_Data); - - #if DEBUG_DV GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "%s: Receives DATA message from %s size %d, packed size %d!\n", my_short_id, GNUNET_i2s(peer) , ntohs(incoming->header.size), packed_message_size); @@ -1258,7 +1397,6 @@ static int handle_dv_data_message (void *cls, if (ntohs (incoming->header.size) < sizeof (p2p_dv_MESSAGE_Data) + sizeof (struct GNUNET_MessageHeader)) { - #if DEBUG_DV GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "`%s': Message sizes don't add up, total size %u, expected at least %u!\n", "dv service", ntohs(incoming->header.size), sizeof (p2p_dv_MESSAGE_Data) + sizeof (struct GNUNET_MessageHeader)); @@ -1266,28 +1404,26 @@ static int handle_dv_data_message (void *cls, return GNUNET_SYSERR; } - dn = GNUNET_CONTAINER_multihashmap_get (ctx.direct_neighbors, - &peer->hashPubKey); + /* Iterate over ATS_Information to get distance and latency */ + latency = get_atsi_latency(atsi); + distance = get_atsi_distance(atsi); + dn = GNUNET_CONTAINER_multihashmap_get (direct_neighbors, + &peer->hashPubKey); if (dn == NULL) - { -#if DEBUG_DV - GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, - "%s: dn NULL!\n", "dv"); -#endif - return GNUNET_OK; - } + return GNUNET_OK; + sid = ntohl (incoming->sender); #if USE_PEER_ID if (sid != 0) { checkPeerCtx.sender_id = sid; checkPeerCtx.peer = NULL; - GNUNET_CONTAINER_multihashmap_iterate(ctx.extended_neighbors, &checkPeerID, &checkPeerCtx); + GNUNET_CONTAINER_multihashmap_iterate(extended_neighbors, &checkPeerID, &checkPeerCtx); pos = checkPeerCtx.peer; } else { - pos = GNUNET_CONTAINER_multihashmap_get (ctx.extended_neighbors, + pos = GNUNET_CONTAINER_multihashmap_get (extended_neighbors, &peer->hashPubKey); } #else @@ -1298,12 +1434,9 @@ static int handle_dv_data_message (void *cls, if (pos == NULL) { - direct_id = GNUNET_strdup(GNUNET_i2s(&dn->identity)); -#if DEBUG_DV +#if DEBUG_DV_MESSAGES GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, - "%s: unknown sender (%u), Message from %s!\n", GNUNET_i2s(&my_identity), ntohl(incoming->sender), direct_id); -#endif - GNUNET_free(direct_id); + "%s: unknown sender (%u), Message uid %u from %s!\n", my_short_id, ntohl(incoming->sender), ntohl(incoming->uid), GNUNET_i2s(&dn->identity)); pos = dn->referee_head; while ((NULL != pos) && (pos->referrer_id != sid)) { @@ -1312,42 +1445,55 @@ static int handle_dv_data_message (void *cls, GNUNET_free(sender_id); pos = pos->next; } - -#if DEBUG_MESSAGE_DROP - direct_id = GNUNET_strdup(GNUNET_i2s(&dn->identity)); - GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, - "%s: DROPPING MESSAGE type %d, unknown sender! Message immediately from %s!\n", GNUNET_i2s(&my_identity), ntohs(((struct GNUNET_MessageHeader *)&incoming[1])->type), direct_id); - GNUNET_free(direct_id); #endif + + found_pos = -1; + for (i = 0; i< MAX_OUTSTANDING_MESSAGES; i++) + { + if (dn->pending_messages[i].sender_id == 0) + { + found_pos = i; + break; + } + } + + if (found_pos == -1) + { + GNUNET_log (GNUNET_ERROR_TYPE_WARNING, + "%s: Too many unknown senders (%u), ignoring message! Message uid %llu from %s!\n", my_short_id, ntohl(incoming->sender), ntohl(incoming->uid), GNUNET_i2s(&dn->identity)); + } + else + { + dn->pending_messages[found_pos].message = GNUNET_malloc(ntohs (message->size)); + memcpy(dn->pending_messages[found_pos].message, message, ntohs(message->size)); + dn->pending_messages[found_pos].distance = distance; + dn->pending_messages[found_pos].latency = latency; + memcpy(&dn->pending_messages[found_pos].sender, peer, sizeof(struct GNUNET_PeerIdentity)); + dn->pending_messages[found_pos].sender_id = sid; + } /* unknown sender */ return GNUNET_OK; } - original_sender = pos->identity; + original_sender = &pos->identity; tid = ntohl (incoming->recipient); if (tid == 0) { /* 0 == us */ cbuf = (char *)&incoming[1]; - offset = 0; - while(offset < packed_message_size) - { - packed_message = (struct GNUNET_MessageHeader *)&cbuf[offset]; -#if DEBUG_DV_MESSAGES - GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, - "%s: Receives %s message for me, uid %u, size %d type %d, cost %u from %s!\n", my_short_id, "DV DATA", ntohl(incoming->uid), ntohs(packed_message->size), ntohs(packed_message->type), pos->cost, GNUNET_i2s(&pos->identity)); -#endif - GNUNET_break_op (ntohs (packed_message->type) != GNUNET_MESSAGE_TYPE_DV_GOSSIP); - GNUNET_break_op (ntohs (packed_message->type) != GNUNET_MESSAGE_TYPE_DV_DATA); - if ( (ntohs (packed_message->type) != GNUNET_MESSAGE_TYPE_DV_GOSSIP) && - (ntohs (packed_message->type) != GNUNET_MESSAGE_TYPE_DV_DATA) ) - { - GNUNET_assert(memcmp(peer, &pos->identity, sizeof(struct GNUNET_PeerIdentity)) != 0); - send_to_plugin(peer, packed_message, ntohs(packed_message->size), &pos->identity, pos->cost); - } - offset += ntohs(packed_message->size); + tkm_ctx.peer = peer; + tkm_ctx.distant = pos; + tkm_ctx.uid = ntohl(incoming->uid); + if (GNUNET_OK != GNUNET_SERVER_mst_receive (coreMST, + &tkm_ctx, + cbuf, + packed_message_size, + GNUNET_NO, + GNUNET_NO)) + { + GNUNET_break_op(0); + GNUNET_log(GNUNET_ERROR_TYPE_WARNING, "%s: %s Received corrupt data, discarding!", my_short_id, "DV SERVICE"); } - return GNUNET_OK; } else @@ -1361,7 +1507,7 @@ static int handle_dv_data_message (void *cls, issue) */ fdc.tid = tid; fdc.dest = NULL; - GNUNET_CONTAINER_heap_iterate (ctx.neighbor_max_heap, + GNUNET_CONTAINER_heap_iterate (neighbor_max_heap, &find_destination, &fdc); #if DEBUG_DV @@ -1375,21 +1521,17 @@ static int handle_dv_data_message (void *cls, GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "%s: Receives %s message uid %u for someone we don't know (id %u)!\n", my_short_id, "DV DATA", ntohl(incoming->uid), tid); #endif - return GNUNET_OK; + return GNUNET_OK; } - destination = fdc.dest->identity; + destination = &fdc.dest->identity; - if (0 == memcmp (&destination, peer, sizeof (struct GNUNET_PeerIdentity))) + if (0 == memcmp (destination, peer, sizeof (struct GNUNET_PeerIdentity))) { /* FIXME: create stat: routing loop-discard! */ -#if DEBUG_DV_PEER_NUMBERS - GNUNET_log(GNUNET_ERROR_TYPE_DEBUG, "\n\n\nLoopy loo message\n\n\n"); -#endif #if DEBUG_DV_MESSAGES - direct_id = GNUNET_strdup(GNUNET_i2s(&dn->identity)); GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, - "%s: DROPPING MESSAGE uid %u type %d, routing loop! Message immediately from %s!\n", my_short_id, ntohl(incoming->uid), ntohs(packed_message->type), direct_id); + "%s: DROPPING MESSAGE uid %u type %d, routing loop! Message immediately from %s!\n", my_short_id, ntohl(incoming->uid), ntohs(packed_message->type), GNUNET_i2s(&dn->identity)); #endif return GNUNET_OK; } @@ -1397,37 +1539,45 @@ static int handle_dv_data_message (void *cls, /* At this point we have a message, and we need to forward it on to the * next DV hop. */ - /* FIXME: Can't send message on, we have to behave. - * We have to tell core we have a message for the next peer, and let - * transport do transport selection on how to get this message to 'em */ - /*ret = send_message (&destination, - &original_sender, - packed_message, DV_PRIORITY, DV_DELAY);*/ - #if DEBUG_DV_MESSAGES GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, - "%s: FORWARD %s message for %s, uid %u, size %d type %d, cost %u!\n", my_short_id, "DV DATA", GNUNET_i2s(&destination), ntohl(incoming->uid), ntohs(packed_message->size), ntohs(packed_message->type), pos->cost); + "%s: FORWARD %s message for %s, uid %u, size %d type %d, cost %u!\n", my_short_id, "DV DATA", GNUNET_i2s(destination), ntohl(incoming->uid), ntohs(packed_message->size), ntohs(packed_message->type), pos->cost); #endif - ret = send_message(&destination, - &original_sender, - NULL, - packed_message, - packed_message_size, - default_dv_priority, -#if DEBUG_DV_MESSAGES - ntohl(incoming->uid), +#if DELAY_FORWARDS + if (GNUNET_TIME_absolute_get_duration(pos->last_gossip).abs_value < GNUNET_TIME_relative_multiply(GNUNET_TIME_UNIT_SECONDS, 2).abs_value) + { + delayed_context = GNUNET_malloc(sizeof(struct DelayedMessageContext)); + memcpy(&delayed_context->dest, destination, sizeof(struct GNUNET_PeerIdentity)); + memcpy(&delayed_context->sender, original_sender, sizeof(struct GNUNET_PeerIdentity)); + delayed_context->message = GNUNET_malloc(packed_message_size); + memcpy(delayed_context->message, packed_message, packed_message_size); + delayed_context->message_size = packed_message_size; + delayed_context->uid = ntohl(incoming->uid); + GNUNET_SCHEDULER_add_delayed(GNUNET_TIME_relative_multiply(GNUNET_TIME_UNIT_MILLISECONDS, 2500), &send_message_delayed, delayed_context); + return GNUNET_OK; + } + else #endif - GNUNET_TIME_relative_get_forever()); - + { + ret = send_message(destination, + original_sender, + NULL, + packed_message, + packed_message_size, + default_dv_priority, + ntohl(incoming->uid), + GNUNET_TIME_relative_get_forever()); + } if (ret != GNUNET_SYSERR) return GNUNET_OK; else { #if DEBUG_MESSAGE_DROP - direct_id = GNUNET_strdup(GNUNET_i2s(&dn->identity)); + char *direct_id = GNUNET_strdup(GNUNET_i2s(&dn->identity)); GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "%s: DROPPING MESSAGE type %d, forwarding failed! Message immediately from %s!\n", GNUNET_i2s(&my_identity), ntohs(((struct GNUNET_MessageHeader *)&incoming[1])->type), direct_id); + GNUNET_free (direct_id); #endif return GNUNET_SYSERR; } @@ -1446,9 +1596,9 @@ static int handle_dv_data_message (void *cls, */ int print_neighbors (void *cls, const GNUNET_HashCode * key, - void *value) + void *abs_value) { - struct DistantNeighbor *distant_neighbor = value; + struct DistantNeighbor *distant_neighbor = abs_value; char my_shortname[5]; char referrer_shortname[5]; memcpy(&my_shortname, GNUNET_i2s(&my_identity), 4); @@ -1481,15 +1631,15 @@ neighbor_send_task (void *cls, p2p_dv_MESSAGE_NeighborInfo *message; struct PendingMessage *pending_message; - if (tc->reason == GNUNET_SCHEDULER_REASON_SHUTDOWN) - { + if ( (tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN) != 0) + { #if DEBUG_DV_GOSSIP GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "%s: Called with reason shutdown, shutting down!\n", GNUNET_i2s(&my_identity)); #endif - return; - } + return; + } if (send_context->fast_gossip_list_head != NULL) { @@ -1514,7 +1664,7 @@ neighbor_send_task (void *cls, * * NOTE: probably fixed once we decided send rate based on allowed bandwidth. */ - about = GNUNET_CONTAINER_heap_walk_get_next (ctx.neighbor_min_heap); + about = GNUNET_CONTAINER_heap_walk_get_next (neighbor_min_heap); } to = send_context->toNeighbor; @@ -1537,6 +1687,7 @@ neighbor_send_task (void *cls, GNUNET_free(encPeerAbout); GNUNET_free(encPeerTo); #endif + about->last_gossip = GNUNET_TIME_absolute_get(); pending_message = GNUNET_malloc(sizeof(struct PendingMessage) + sizeof(p2p_dv_MESSAGE_NeighborInfo)); pending_message->msg = (struct GNUNET_MessageHeader *)&pending_message[1]; pending_message->importance = default_dv_priority; @@ -1558,8 +1709,9 @@ neighbor_send_task (void *cls, core_pending_tail, pending_message); - if (core_transmit_handle == NULL) - core_transmit_handle = GNUNET_CORE_notify_transmit_ready(coreAPI, default_dv_priority, GNUNET_TIME_relative_get_forever(), &to->identity, sizeof(p2p_dv_MESSAGE_NeighborInfo), &core_transmit_notify, NULL); + GNUNET_SCHEDULER_add_now(try_core_send, NULL); + /*if (core_transmit_handle == NULL) + core_transmit_handle = GNUNET_CORE_notify_transmit_ready(coreAPI, GNUNET_YES, default_dv_priority, GNUNET_TIME_relative_get_forever(), &to->identity, sizeof(p2p_dv_MESSAGE_NeighborInfo), &core_transmit_notify, NULL);*/ } @@ -1568,14 +1720,14 @@ neighbor_send_task (void *cls, #if DEBUG_DV_PEER_NUMBERS GNUNET_log(GNUNET_ERROR_TYPE_WARNING, "DV SERVICE: still in fast send mode\n"); #endif - send_context->task = GNUNET_SCHEDULER_add_now(sched, &neighbor_send_task, send_context); + send_context->task = GNUNET_SCHEDULER_add_now(&neighbor_send_task, send_context); } else { #if DEBUG_DV_PEER_NUMBERS GNUNET_log(GNUNET_ERROR_TYPE_WARNING, "DV SERVICE: entering slow send mode\n"); #endif - send_context->task = GNUNET_SCHEDULER_add_delayed(sched, GNUNET_DV_DEFAULT_SEND_INTERVAL, &neighbor_send_task, send_context); + send_context->task = GNUNET_SCHEDULER_add_delayed(GNUNET_DV_DEFAULT_SEND_INTERVAL, &neighbor_send_task, send_context); } return; @@ -1621,10 +1773,10 @@ handle_start (void *cls, */ int send_iterator (void *cls, const GNUNET_HashCode * key, - void *value) + void *abs_value) { struct DV_SendContext *send_context = cls; - struct DistantNeighbor *distant_neighbor = value; + struct DistantNeighbor *distant_neighbor = abs_value; if (memcmp(distant_neighbor->referrer, send_context->direct_peer, sizeof(struct GNUNET_PeerIdentity)) == 0) /* They match, send and free */ { @@ -1685,9 +1837,7 @@ void handle_dv_send_message (void *cls, address_len = ntohl(send_msg->addrlen); GNUNET_assert(address_len == sizeof(struct GNUNET_PeerIdentity) * 2); - message_size = ntohl(send_msg->msgbuf_size); - - GNUNET_assert(ntohs(message->size) == sizeof(struct GNUNET_DV_SendMessage) + address_len + message_size); + message_size = ntohs(message->size) - sizeof(struct GNUNET_DV_SendMessage) - address_len; destination = GNUNET_malloc(sizeof(struct GNUNET_PeerIdentity)); direct = GNUNET_malloc(sizeof(struct GNUNET_PeerIdentity)); message_buf = GNUNET_malloc(message_size); @@ -1772,7 +1922,7 @@ void handle_dv_send_message (void *cls, /* In bizarro world GNUNET_SYSERR indicates that we succeeded */ #if UNSIMPLER - if (GNUNET_SYSERR != GNUNET_CONTAINER_multihashmap_get_multiple(ctx.extended_neighbors, &destination->hashPubKey, &send_iterator, send_context)) + if (GNUNET_SYSERR != GNUNET_CONTAINER_multihashmap_get_multiple(extended_neighbors, &destination->hashPubKey, &send_iterator, send_context)) { send_result_msg->result = htons(1); pending_message = GNUNET_malloc(sizeof(struct PendingMessage) + sizeof(struct GNUNET_DV_SendResultMessage)); @@ -1813,14 +1963,12 @@ void handle_dv_send_message (void *cls, static int handle_dv_gossip_message (void *cls, const struct GNUNET_PeerIdentity *peer, const struct GNUNET_MessageHeader *message, - struct GNUNET_TIME_Relative latency, - uint32_t distance); + const struct GNUNET_TRANSPORT_ATS_Information *atsi); static int handle_dv_disconnect_message (void *cls, const struct GNUNET_PeerIdentity *peer, const struct GNUNET_MessageHeader *message, - struct GNUNET_TIME_Relative latency, - uint32_t distance); + const struct GNUNET_TRANSPORT_ATS_Information *atsi); /** End forward declarations **/ @@ -1862,9 +2010,9 @@ distant_neighbor_free (struct DistantNeighbor *referee) GNUNET_CONTAINER_DLL_remove (referrer->referee_head, referrer->referee_tail, referee); } - GNUNET_CONTAINER_heap_remove_node (ctx.neighbor_max_heap, referee->max_loc); - GNUNET_CONTAINER_heap_remove_node (ctx.neighbor_min_heap, referee->min_loc); - GNUNET_CONTAINER_multihashmap_remove_all (ctx.extended_neighbors, + GNUNET_CONTAINER_heap_remove_node (referee->max_loc); + GNUNET_CONTAINER_heap_remove_node (referee->min_loc); + GNUNET_CONTAINER_multihashmap_remove_all (extended_neighbors, &referee->identity.hashPubKey); GNUNET_free_non_null (referee->pkey); GNUNET_free (referee); @@ -1884,7 +2032,7 @@ direct_neighbor_free (struct DirectNeighbor *direct) send_context = direct->send_context; if (send_context->task != GNUNET_SCHEDULER_NO_TASK) - GNUNET_SCHEDULER_cancel(sched, send_context->task); + GNUNET_SCHEDULER_cancel(send_context->task); about_list = send_context->fast_gossip_list_head; while (about_list != NULL) @@ -1937,8 +2085,9 @@ static int schedule_disconnect_messages (void *cls, core_pending_tail, pending_message); - if (core_transmit_handle == NULL) - core_transmit_handle = GNUNET_CORE_notify_transmit_ready(coreAPI, default_dv_priority, GNUNET_TIME_relative_get_forever(), ¬ify->identity, sizeof(p2p_dv_MESSAGE_Disconnect), &core_transmit_notify, NULL); + GNUNET_SCHEDULER_add_now(try_core_send, NULL); + /*if (core_transmit_handle == NULL) + core_transmit_handle = GNUNET_CORE_notify_transmit_ready(coreAPI, GNUNET_YES, default_dv_priority, GNUNET_TIME_relative_get_forever(), ¬ify->identity, sizeof(p2p_dv_MESSAGE_Disconnect), &core_transmit_notify, NULL);*/ return GNUNET_YES; } @@ -1992,19 +2141,20 @@ shutdown_task (void *cls, { #if DEBUG_DV GNUNET_log(GNUNET_ERROR_TYPE_DEBUG, "calling CORE_DISCONNECT\n"); - GNUNET_CONTAINER_multihashmap_iterate(ctx.extended_neighbors, &print_neighbors, NULL); + GNUNET_CONTAINER_multihashmap_iterate(extended_neighbors, &print_neighbors, NULL); #endif - GNUNET_CONTAINER_multihashmap_iterate(ctx.extended_neighbors, &free_extended_neighbors, NULL); - GNUNET_CONTAINER_multihashmap_destroy(ctx.extended_neighbors); - GNUNET_CONTAINER_multihashmap_iterate(ctx.direct_neighbors, &free_direct_neighbors, NULL); - GNUNET_CONTAINER_multihashmap_destroy(ctx.direct_neighbors); + GNUNET_CONTAINER_multihashmap_iterate(extended_neighbors, &free_extended_neighbors, NULL); + GNUNET_CONTAINER_multihashmap_destroy(extended_neighbors); + GNUNET_CONTAINER_multihashmap_iterate(direct_neighbors, &free_direct_neighbors, NULL); + GNUNET_CONTAINER_multihashmap_destroy(direct_neighbors); - GNUNET_CONTAINER_heap_destroy(ctx.neighbor_max_heap); - GNUNET_CONTAINER_heap_destroy(ctx.neighbor_min_heap); + GNUNET_CONTAINER_heap_destroy(neighbor_max_heap); + GNUNET_CONTAINER_heap_destroy(neighbor_min_heap); GNUNET_CORE_disconnect (coreAPI); + coreAPI = NULL; GNUNET_PEERINFO_disconnect(peerinfo_handle); - + GNUNET_SERVER_mst_destroy(coreMST); GNUNET_free_non_null(my_short_id); #if DEBUG_DV GNUNET_log(GNUNET_ERROR_TYPE_DEBUG, "CORE_DISCONNECT completed\n"); @@ -2022,8 +2172,8 @@ void core_init (void *cls, if (server == NULL) { - GNUNET_SCHEDULER_cancel(sched, cleanup_task); - GNUNET_SCHEDULER_add_now(sched, &shutdown_task, NULL); + GNUNET_SCHEDULER_cancel(cleanup_task); + GNUNET_SCHEDULER_add_now(&shutdown_task, NULL); return; } #if DEBUG_DV @@ -2049,10 +2199,10 @@ void core_init (void *cls, */ static int add_pkey_to_extended (void *cls, const GNUNET_HashCode * key, - void *value) + void *abs_value) { struct GNUNET_CRYPTO_RsaPublicKeyBinaryEncoded *pkey = cls; - struct DistantNeighbor *distant_neighbor = value; + struct DistantNeighbor *distant_neighbor = abs_value; if (distant_neighbor->pkey == NULL) { @@ -2084,9 +2234,9 @@ static int update_matching_neighbors (void *cls, if (update_info->referrer == distant_neighbor->referrer) /* Direct neighbor matches, update it's info and return GNUNET_NO */ { /* same referrer, cost change! */ - GNUNET_CONTAINER_heap_update_cost (ctx.neighbor_max_heap, + GNUNET_CONTAINER_heap_update_cost (neighbor_max_heap, update_info->neighbor->max_loc, update_info->cost); - GNUNET_CONTAINER_heap_update_cost (ctx.neighbor_min_heap, + GNUNET_CONTAINER_heap_update_cost (neighbor_min_heap, update_info->neighbor->min_loc, update_info->cost); update_info->neighbor->last_activity = update_info->now; update_info->neighbor->cost = update_info->cost; @@ -2154,12 +2304,57 @@ static int add_distant_all_direct_neighbors (void *cls, GNUNET_free(encPeerTo); #endif /*if (send_context->task != GNUNET_SCHEDULER_NO_TASK) - GNUNET_SCHEDULER_cancel(sched, send_context->task);*/ + GNUNET_SCHEDULER_cancel(send_context->task);*/ - send_context->task = GNUNET_SCHEDULER_add_now(sched, &neighbor_send_task, send_context); + send_context->task = GNUNET_SCHEDULER_add_now(&neighbor_send_task, send_context); return GNUNET_YES; } +/** + * Callback for hello address creation. + * + * @param cls closure, a struct HelloContext + * @param max maximum number of bytes that can be written to buf + * @param buf where to write the address information + * + * @return number of bytes written, 0 to signal the + * end of the iteration. + */ +static size_t +generate_hello_address (void *cls, size_t max, void *buf) +{ + struct HelloContext *hello_context = cls; + char *addr_buffer; + size_t offset; + size_t size; + size_t ret; + + if (hello_context->addresses_to_add == 0) + return 0; + + /* Hello "address" will be concatenation of distant peer and direct peer identities */ + size = 2 * sizeof(struct GNUNET_PeerIdentity); + GNUNET_assert(max >= size); + + addr_buffer = GNUNET_malloc(size); + offset = 0; + /* Copy the distant peer identity to buffer */ + memcpy(addr_buffer, &hello_context->distant_peer, sizeof(struct GNUNET_PeerIdentity)); + offset += sizeof(struct GNUNET_PeerIdentity); + /* Copy the direct peer identity to buffer */ + memcpy(&addr_buffer[offset], hello_context->direct_peer, sizeof(struct GNUNET_PeerIdentity)); + ret = GNUNET_HELLO_add_address ("dv", + GNUNET_TIME_relative_to_absolute + (GNUNET_TIME_UNIT_HOURS), addr_buffer, size, + buf, max); + + hello_context->addresses_to_add--; + + GNUNET_free(addr_buffer); + return ret; +} + + /** * Handles when a peer is either added due to being newly connected * or having been gossiped about, also called when the cost for a neighbor @@ -2175,7 +2370,8 @@ static int add_distant_all_direct_neighbors (void *cls, * not added) */ static struct DistantNeighbor * -addUpdateNeighbor (const struct GNUNET_PeerIdentity * peer, struct GNUNET_CRYPTO_RsaPublicKeyBinaryEncoded *pkey, +addUpdateNeighbor (const struct GNUNET_PeerIdentity * peer, + struct GNUNET_CRYPTO_RsaPublicKeyBinaryEncoded *pkey, unsigned int referrer_peer_id, struct DirectNeighbor *referrer, unsigned int cost) { @@ -2183,7 +2379,12 @@ addUpdateNeighbor (const struct GNUNET_PeerIdentity * peer, struct GNUNET_CRYPTO struct DistantNeighbor *max; struct GNUNET_TIME_Absolute now; struct NeighborUpdateInfo *neighbor_update; + struct HelloContext *hello_context; + struct GNUNET_HELLO_Message *hello_msg; unsigned int our_id; + char *addr1; + char *addr2; + int i; #if DEBUG_DV_PEER_NUMBERS char *encAbout; @@ -2192,7 +2393,7 @@ addUpdateNeighbor (const struct GNUNET_PeerIdentity * peer, struct GNUNET_CRYPTO #endif now = GNUNET_TIME_absolute_get (); - neighbor = GNUNET_CONTAINER_multihashmap_get (ctx.extended_neighbors, + neighbor = GNUNET_CONTAINER_multihashmap_get (extended_neighbors, &peer->hashPubKey); neighbor_update = GNUNET_malloc(sizeof(struct NeighborUpdateInfo)); neighbor_update->neighbor = neighbor; @@ -2220,7 +2421,7 @@ addUpdateNeighbor (const struct GNUNET_PeerIdentity * peer, struct GNUNET_CRYPTO /* Either we do not know this peer, or we already do but via a different immediate peer */ if ((neighbor == NULL) || - (GNUNET_CONTAINER_multihashmap_get_multiple(ctx.extended_neighbors, + (GNUNET_CONTAINER_multihashmap_get_multiple(extended_neighbors, &peer->hashPubKey, &update_matching_neighbors, neighbor_update) != GNUNET_SYSERR)) @@ -2231,13 +2432,13 @@ addUpdateNeighbor (const struct GNUNET_PeerIdentity * peer, struct GNUNET_CRYPTO { distant_neighbor_free(neighbor); } - else if ((neighbor != NULL) && (cost >= neighbor->cost)) /* Only allow one DV connection to each peer */ + else if (neighbor != NULL) /* Only allow one DV connection to each peer */ { return NULL; } #endif /* new neighbor! */ - if (cost > ctx.fisheye_depth) + if (cost > fisheye_depth) { /* too costly */ GNUNET_free(neighbor_update); @@ -2251,11 +2452,12 @@ addUpdateNeighbor (const struct GNUNET_PeerIdentity * peer, struct GNUNET_CRYPTO GNUNET_free(encAbout); #endif - if (ctx.max_table_size <= - GNUNET_CONTAINER_multihashmap_size (ctx.extended_neighbors)) + if (max_table_size <= + GNUNET_CONTAINER_multihashmap_size (extended_neighbors)) { /* remove most expensive entry */ - max = GNUNET_CONTAINER_heap_peek (ctx.neighbor_max_heap); + max = GNUNET_CONTAINER_heap_peek (neighbor_max_heap); + GNUNET_assert(max != NULL); if (cost > max->cost) { /* new entry most expensive, don't create */ @@ -2274,9 +2476,9 @@ addUpdateNeighbor (const struct GNUNET_PeerIdentity * peer, struct GNUNET_CRYPTO neighbor = GNUNET_malloc (sizeof (struct DistantNeighbor)); GNUNET_CONTAINER_DLL_insert (referrer->referee_head, referrer->referee_tail, neighbor); - neighbor->max_loc = GNUNET_CONTAINER_heap_insert (ctx.neighbor_max_heap, + neighbor->max_loc = GNUNET_CONTAINER_heap_insert (neighbor_max_heap, neighbor, cost); - neighbor->min_loc = GNUNET_CONTAINER_heap_insert (ctx.neighbor_min_heap, + neighbor->min_loc = GNUNET_CONTAINER_heap_insert (neighbor_min_heap, neighbor, cost); neighbor->referrer = referrer; memcpy (&neighbor->identity, peer, sizeof (struct GNUNET_PeerIdentity)); @@ -2296,9 +2498,54 @@ addUpdateNeighbor (const struct GNUNET_PeerIdentity * peer, struct GNUNET_CRYPTO (cost == DIRECT_NEIGHBOR_COST) ? (GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK, 4) == 0) : GNUNET_NO; - GNUNET_CONTAINER_multihashmap_put (ctx.extended_neighbors, &peer->hashPubKey, + GNUNET_CONTAINER_multihashmap_put (extended_neighbors, &peer->hashPubKey, neighbor, GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE); + if (referrer_peer_id != 0) + { + for (i = 0; i< MAX_OUTSTANDING_MESSAGES; i++) + { + if (referrer->pending_messages[i].sender_id == referrer_peer_id) /* We have a queued message from just learned about peer! */ + { +#if DEBUG_DV_MESSAGES + GNUNET_log(GNUNET_ERROR_TYPE_DEBUG, "%s: learned about peer %llu from which we have a previous unknown message, processing!\n", my_short_id, referrer_peer_id); +#endif + struct GNUNET_TRANSPORT_ATS_Information atsi[3]; + atsi[0].type = htonl (GNUNET_TRANSPORT_ATS_QUALITY_NET_DISTANCE); + atsi[0].value = htonl (referrer->pending_messages[i].distance); + atsi[1].type = htonl (GNUNET_TRANSPORT_ATS_QUALITY_NET_DELAY); + atsi[1].value = htonl ((uint32_t)referrer->pending_messages[i].latency.rel_value); + atsi[2].type = htonl (GNUNET_TRANSPORT_ATS_ARRAY_TERMINATOR); + atsi[2].value = htonl (0); + handle_dv_data_message(NULL, + &referrer->pending_messages[i].sender, + referrer->pending_messages[i].message, + (const struct GNUNET_TRANSPORT_ATS_Information *)&atsi); + GNUNET_free(referrer->pending_messages[i].message); + referrer->pending_messages[i].sender_id = 0; + } + } + } + if ((cost != DIRECT_NEIGHBOR_COST) && (neighbor->pkey != NULL)) + { + /* Added neighbor, now send HELLO to transport */ + hello_context = GNUNET_malloc(sizeof(struct HelloContext)); + hello_context->direct_peer = &referrer->identity; + memcpy(&hello_context->distant_peer, peer, sizeof(struct GNUNET_PeerIdentity)); + hello_context->addresses_to_add = 1; + hello_msg = GNUNET_HELLO_create(pkey, &generate_hello_address, hello_context); + GNUNET_assert(memcmp(hello_context->direct_peer, &hello_context->distant_peer, sizeof(struct GNUNET_PeerIdentity)) != 0); + addr1 = GNUNET_strdup(GNUNET_i2s(hello_context->direct_peer)); + addr2 = GNUNET_strdup(GNUNET_i2s(&hello_context->distant_peer)); +#if DEBUG_DV + GNUNET_log(GNUNET_ERROR_TYPE_DEBUG, "%s: GIVING HELLO size %d for %s via %s to TRANSPORT\n", my_short_id, GNUNET_HELLO_size(hello_msg), addr2, addr1); +#endif + GNUNET_free(addr1); + GNUNET_free(addr2); + send_to_plugin(hello_context->direct_peer, GNUNET_HELLO_get_header(hello_msg), GNUNET_HELLO_size(hello_msg), &hello_context->distant_peer, cost); + GNUNET_free(hello_context); + GNUNET_free(hello_msg); + } } else @@ -2310,7 +2557,7 @@ addUpdateNeighbor (const struct GNUNET_PeerIdentity * peer, struct GNUNET_CRYPTO } #if DEBUG_DV GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, - "%s: Size of extended_neighbors is %d\n", "dv", GNUNET_CONTAINER_multihashmap_size(ctx.extended_neighbors)); + "%s: Size of extended_neighbors is %d\n", "dv", GNUNET_CONTAINER_multihashmap_size(extended_neighbors)); #endif GNUNET_free(neighbor_update); @@ -2318,48 +2565,6 @@ addUpdateNeighbor (const struct GNUNET_PeerIdentity * peer, struct GNUNET_CRYPTO } -static size_t -generate_hello_address (void *cls, size_t max, void *buf) -{ - struct HelloContext *hello_context = cls; - char *addr_buffer; - size_t offset; - size_t size; - size_t ret; - char *addr1; - char *addr2; - - if (hello_context->addresses_to_add == 0) - return 0; - - /* Hello "address" will be concatenation of distant peer and direct peer identities */ - size = 2 * sizeof(struct GNUNET_PeerIdentity); - GNUNET_assert(max >= size); - - addr_buffer = GNUNET_malloc(size); - offset = 0; - /* Copy the distant peer identity to buffer */ - memcpy(addr_buffer, &hello_context->distant_peer, sizeof(struct GNUNET_PeerIdentity)); - offset += sizeof(struct GNUNET_PeerIdentity); - /* Copy the direct peer identity to buffer */ - memcpy(&addr_buffer[offset], hello_context->direct_peer, sizeof(struct GNUNET_PeerIdentity)); - addr1 = GNUNET_strdup(GNUNET_i2s(hello_context->direct_peer)); - addr2 = GNUNET_strdup(GNUNET_i2s(&hello_context->distant_peer)); - GNUNET_log(GNUNET_ERROR_TYPE_DEBUG, "%s: GIVING HELLO %s%s%s to TRANSPORT\n", my_short_id,my_short_id, addr1, addr2); - GNUNET_free(addr1); - GNUNET_free(addr2); - ret = GNUNET_HELLO_add_address ("dv", - GNUNET_TIME_relative_to_absolute - (GNUNET_TIME_UNIT_HOURS), addr_buffer, size, - buf, max); - - hello_context->addresses_to_add--; - - GNUNET_free(addr_buffer); - return ret; -} - - /** * Core handler for dv disconnect messages. These will be used * by us to tell transport via the dv plugin that a peer can @@ -2370,14 +2575,12 @@ generate_hello_address (void *cls, size_t max, void *buf) * @param cls closure * @param peer peer which sent the message (immediate sender) * @param message the message - * @param latency the latency of the connection we received the message from - * @param distance the distance to the immediate peer + * @param atsi performance data */ static int handle_dv_disconnect_message (void *cls, const struct GNUNET_PeerIdentity *peer, const struct GNUNET_MessageHeader *message, - struct GNUNET_TIME_Relative latency, - uint32_t distance) + const struct GNUNET_TRANSPORT_ATS_Information *atsi) { struct DirectNeighbor *referrer; struct DistantNeighbor *distant; @@ -2388,7 +2591,7 @@ static int handle_dv_disconnect_message (void *cls, return GNUNET_SYSERR; /* invalid message */ } - referrer = GNUNET_CONTAINER_multihashmap_get (ctx.direct_neighbors, + referrer = GNUNET_CONTAINER_multihashmap_get (direct_neighbors, &peer->hashPubKey); if (referrer == NULL) return GNUNET_OK; @@ -2399,8 +2602,10 @@ static int handle_dv_disconnect_message (void *cls, if (distant->referrer_id == ntohl(enc_message->peer_id)) { distant_neighbor_free(distant); + distant = referrer->referee_head; } - distant = referrer->referee_head; + else + distant = distant->next; } return GNUNET_OK; @@ -2417,17 +2622,14 @@ static int handle_dv_disconnect_message (void *cls, * @param cls closure * @param peer peer which sent the message (immediate sender) * @param message the message - * @param latency the latency of the connection we received the message from - * @param distance the distance to the immediate peer + * @param atsi performance data */ -static int handle_dv_gossip_message (void *cls, - const struct GNUNET_PeerIdentity *peer, - const struct GNUNET_MessageHeader *message, - struct GNUNET_TIME_Relative latency, - uint32_t distance) +static int +handle_dv_gossip_message (void *cls, + const struct GNUNET_PeerIdentity *peer, + const struct GNUNET_MessageHeader *message, + const struct GNUNET_TRANSPORT_ATS_Information *atsi) { - struct HelloContext *hello_context; - struct GNUNET_HELLO_Message *hello_msg; struct DirectNeighbor *referrer; p2p_dv_MESSAGE_NeighborInfo *enc_message = (p2p_dv_MESSAGE_NeighborInfo *)message; @@ -2448,7 +2650,7 @@ static int handle_dv_gossip_message (void *cls, GNUNET_free(encPeerFrom); #endif - referrer = GNUNET_CONTAINER_multihashmap_get (ctx.direct_neighbors, + referrer = GNUNET_CONTAINER_multihashmap_get (direct_neighbors, &peer->hashPubKey); if (referrer == NULL) return GNUNET_OK; @@ -2457,15 +2659,6 @@ static int handle_dv_gossip_message (void *cls, ntohl (enc_message->neighbor_id), referrer, ntohl (enc_message->cost) + 1); - hello_context = GNUNET_malloc(sizeof(struct HelloContext)); - hello_context->direct_peer = peer; - memcpy(&hello_context->distant_peer, &enc_message->neighbor, sizeof(struct GNUNET_PeerIdentity)); - hello_context->addresses_to_add = 1; - hello_msg = GNUNET_HELLO_create(&enc_message->pkey, &generate_hello_address, hello_context); - GNUNET_assert(memcmp(hello_context->direct_peer, &hello_context->distant_peer, sizeof(struct GNUNET_PeerIdentity)) != 0); - send_to_plugin(hello_context->direct_peer, GNUNET_HELLO_get_header(hello_msg), GNUNET_HELLO_size(hello_msg), &hello_context->distant_peer, ntohl(enc_message->cost) + 1); - GNUNET_free(hello_context); - GNUNET_free(hello_msg); return GNUNET_OK; } @@ -2481,9 +2674,10 @@ static int handle_dv_gossip_message (void *cls, * * @return GNUNET_YES to continue iteration, GNUNET_NO otherwise */ -static int add_all_extended_peers (void *cls, - const GNUNET_HashCode * key, - void *value) +static int +add_all_extended_peers (void *cls, + const GNUNET_HashCode * key, + void *value) { struct NeighborSendContext *send_context = (struct NeighborSendContext *)cls; struct DistantNeighbor *distant = (struct DistantNeighbor *)value; @@ -2518,18 +2712,19 @@ static int add_all_extended_peers (void *cls, * iterate, * GNUNET_NO if not. */ -static int gossip_all_to_all_iterator (void *cls, - const GNUNET_HashCode * key, - void *value) +static int +gossip_all_to_all_iterator (void *cls, + const GNUNET_HashCode * key, + void *abs_value) { - struct DirectNeighbor *direct = value; + struct DirectNeighbor *direct = abs_value; - GNUNET_CONTAINER_multihashmap_iterate (ctx.extended_neighbors, &add_all_extended_peers, direct->send_context); + GNUNET_CONTAINER_multihashmap_iterate (extended_neighbors, &add_all_extended_peers, direct->send_context); if (direct->send_context->task != GNUNET_SCHEDULER_NO_TASK) - GNUNET_SCHEDULER_cancel(sched, direct->send_context->task); + GNUNET_SCHEDULER_cancel(direct->send_context->task); - direct->send_context->task = GNUNET_SCHEDULER_add_now(sched, &neighbor_send_task, direct->send_context); + direct->send_context->task = GNUNET_SCHEDULER_add_now(&neighbor_send_task, direct->send_context); return GNUNET_YES; } @@ -2543,10 +2738,9 @@ static void gossip_all_to_all (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc) { - GNUNET_CONTAINER_multihashmap_iterate (ctx.direct_neighbors, &gossip_all_to_all_iterator, NULL); + GNUNET_CONTAINER_multihashmap_iterate (direct_neighbors, &gossip_all_to_all_iterator, NULL); - GNUNET_SCHEDULER_add_delayed (sched, - GNUNET_TIME_relative_multiply(GNUNET_TIME_UNIT_SECONDS, 5), + GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_relative_multiply(GNUNET_TIME_UNIT_SECONDS, 5), &gossip_all_to_all, NULL); @@ -2563,9 +2757,10 @@ gossip_all_to_all (void *cls, * * @return GNUNET_YES to continue iteration, GNUNET_NO otherwise */ -static int add_all_direct_neighbors (void *cls, - const GNUNET_HashCode * key, - void *value) +static int +add_all_direct_neighbors (void *cls, + const GNUNET_HashCode * key, + void *value) { struct DirectNeighbor *direct = (struct DirectNeighbor *)value; struct DirectNeighbor *to = (struct DirectNeighbor *)cls; @@ -2575,7 +2770,7 @@ static int add_all_direct_neighbors (void *cls, char *direct_id; - distant = GNUNET_CONTAINER_multihashmap_get(ctx.extended_neighbors, &to->identity.hashPubKey); + distant = GNUNET_CONTAINER_multihashmap_get(extended_neighbors, &to->identity.hashPubKey); if (distant == NULL) { return GNUNET_YES; @@ -2603,9 +2798,9 @@ static int add_all_direct_neighbors (void *cls, send_context->fast_gossip_list_tail, gossip_entry); if (send_context->task != GNUNET_SCHEDULER_NO_TASK) - GNUNET_SCHEDULER_cancel(sched, send_context->task); + GNUNET_SCHEDULER_cancel(send_context->task); - send_context->task = GNUNET_SCHEDULER_add_now(sched, &neighbor_send_task, send_context); + send_context->task = GNUNET_SCHEDULER_add_now(&neighbor_send_task, send_context); //tc.reason = GNUNET_SCHEDULER_REASON_TIMEOUT; //neighbor_send_task(send_context, &tc); return GNUNET_YES; @@ -2618,12 +2813,13 @@ static int add_all_direct_neighbors (void *cls, * @param cls closure * @param peer id of the peer, NULL for last call * @param hello hello message for the peer (can be NULL) - * @param trust amount of trust we have in the peer + * @param err_msg NULL if successful, otherwise contains error message */ static void process_peerinfo (void *cls, const struct GNUNET_PeerIdentity *peer, - const struct GNUNET_HELLO_Message *hello, uint32_t trust) + const struct GNUNET_HELLO_Message *hello, + const char *err_msg) { struct PeerIteratorContext *peerinfo_iterator = cls; struct DirectNeighbor *neighbor = peerinfo_iterator->neighbor; @@ -2632,17 +2828,21 @@ process_peerinfo (void *cls, char *neighbor_pid; #endif int sent; - - if (peer == NULL) /* && (neighbor->pkey == NULL))*/ + if (err_msg != NULL) + { + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, + _("Error in communication with PEERINFO service\n")); + /* return; */ + } + if (peer == NULL) { - if (distant->pkey == NULL) /* FIXME: Reschedule? */ + if (distant->pkey == NULL) { #if DEBUG_DV GNUNET_log(GNUNET_ERROR_TYPE_WARNING, "Failed to get peerinfo information for this peer, retrying!\n"); #endif peerinfo_iterator->ic = GNUNET_PEERINFO_iterate(peerinfo_handle, &peerinfo_iterator->neighbor->identity, - 0, GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 3), &process_peerinfo, peerinfo_iterator); @@ -2665,24 +2865,25 @@ process_peerinfo (void *cls, memcpy(distant->pkey, &neighbor->pkey, sizeof(struct GNUNET_CRYPTO_RsaPublicKeyBinaryEncoded)); } - /* Why do it this way, now we have the distant neighbor! */ - /*GNUNET_CONTAINER_multihashmap_get_multiple(ctx.extended_neighbors, - &peer->hashPubKey, - &add_pkey_to_extended, - &neighbor->pkey);*/ - - sent = GNUNET_CONTAINER_multihashmap_iterate (ctx.extended_neighbors, &add_all_extended_peers, neighbor->send_context); - + sent = GNUNET_CONTAINER_multihashmap_iterate (extended_neighbors, &add_all_extended_peers, neighbor->send_context); + if (stats != NULL) + { + GNUNET_STATISTICS_update (stats, "# distant peers gossiped to direct neighbors", sent, GNUNET_NO); + } #if DEBUG_DV_PEER_NUMBERS neighbor_pid = GNUNET_strdup(GNUNET_i2s(&neighbor->identity)); GNUNET_log(GNUNET_ERROR_TYPE_DEBUG, "%s: Gossipped %d extended peers to %s\n", GNUNET_i2s(&my_identity), sent, neighbor_pid); #endif - sent = GNUNET_CONTAINER_multihashmap_iterate (ctx.direct_neighbors, &add_all_direct_neighbors, neighbor); + sent = GNUNET_CONTAINER_multihashmap_iterate (direct_neighbors, &add_all_direct_neighbors, neighbor); + if (stats != NULL) + { + GNUNET_STATISTICS_update (stats, "# direct peers gossiped to direct neighbors", sent, GNUNET_NO); + } #if DEBUG_DV_PEER_NUMBERS GNUNET_log(GNUNET_ERROR_TYPE_DEBUG, "%s: Gossipped about %s to %d direct peers\n", GNUNET_i2s(&my_identity), neighbor_pid, sent); GNUNET_free(neighbor_pid); #endif - neighbor->send_context->task = GNUNET_SCHEDULER_add_now(sched, &neighbor_send_task, neighbor->send_context); + neighbor->send_context->task = GNUNET_SCHEDULER_add_now(&neighbor_send_task, neighbor->send_context); } } @@ -2692,13 +2893,12 @@ process_peerinfo (void *cls, * * @param cls closure * @param peer peer identity this notification is about - * @param latency reported latency of the connection with peer - * @param distance reported distance (DV) to peer + * @param atsi performance data */ -void handle_core_connect (void *cls, - const struct GNUNET_PeerIdentity * peer, - struct GNUNET_TIME_Relative latency, - uint32_t distance) +static void +handle_core_connect (void *cls, + const struct GNUNET_PeerIdentity * peer, + const struct GNUNET_TRANSPORT_ATS_Information *atsi) { struct DirectNeighbor *neighbor; struct DistantNeighbor *about; @@ -2708,8 +2908,15 @@ void handle_core_connect (void *cls, GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "%s: Receives core connect message for peer %s distance %d!\n", "dv", GNUNET_i2s(peer), distance); #endif + uint32_t distance; + + /* Check for connect to self message */ + if (0 == memcmp(&my_identity, peer, sizeof(struct GNUNET_PeerIdentity))) + return; - if ((distance == DIRECT_NEIGHBOR_COST) && (GNUNET_CONTAINER_multihashmap_get(ctx.direct_neighbors, &peer->hashPubKey) == NULL)) + distance = get_atsi_distance (atsi); + if ((distance == DIRECT_NEIGHBOR_COST) && + (GNUNET_CONTAINER_multihashmap_get(direct_neighbors, &peer->hashPubKey) == NULL)) { peerinfo_iterator = GNUNET_malloc(sizeof(struct PeerIteratorContext)); neighbor = GNUNET_malloc (sizeof (struct DirectNeighbor)); @@ -2717,7 +2924,7 @@ void handle_core_connect (void *cls, neighbor->send_context->toNeighbor = neighbor; memcpy (&neighbor->identity, peer, sizeof (struct GNUNET_PeerIdentity)); - GNUNET_assert(GNUNET_SYSERR != GNUNET_CONTAINER_multihashmap_put (ctx.direct_neighbors, + GNUNET_assert(GNUNET_SYSERR != GNUNET_CONTAINER_multihashmap_put (direct_neighbors, &peer->hashPubKey, neighbor, GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY)); about = addUpdateNeighbor (peer, NULL, 0, neighbor, DIRECT_NEIGHBOR_COST); @@ -2725,7 +2932,6 @@ void handle_core_connect (void *cls, peerinfo_iterator->neighbor = neighbor; peerinfo_iterator->ic = GNUNET_PEERINFO_iterate (peerinfo_handle, peer, - 0, GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 3), &process_peerinfo, peerinfo_iterator); @@ -2743,9 +2949,15 @@ void handle_core_connect (void *cls, } else { - about = GNUNET_CONTAINER_multihashmap_get(ctx.extended_neighbors, &peer->hashPubKey); - if ((GNUNET_CONTAINER_multihashmap_get(ctx.direct_neighbors, &peer->hashPubKey) == NULL) && (about != NULL)) - sent = GNUNET_CONTAINER_multihashmap_iterate(ctx.direct_neighbors, &add_distant_all_direct_neighbors, about); + about = GNUNET_CONTAINER_multihashmap_get(extended_neighbors, &peer->hashPubKey); + if ((GNUNET_CONTAINER_multihashmap_get(direct_neighbors, &peer->hashPubKey) == NULL) && (about != NULL)) + { + sent = GNUNET_CONTAINER_multihashmap_iterate(direct_neighbors, &add_distant_all_direct_neighbors, about); + if (stats != NULL) + { + GNUNET_STATISTICS_update (stats, "# direct peers gossiped to new direct neighbors", sent, GNUNET_NO); + } + } #if DEBUG_DV GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "%s: Distance (%d) greater than %d or already know about peer (%s), not re-adding!\n", "dv", distance, DIRECT_NEIGHBOR_COST, GNUNET_i2s(peer)); @@ -2767,41 +2979,61 @@ void handle_core_disconnect (void *cls, struct DistantNeighbor *referee; struct FindDestinationContext fdc; struct DisconnectContext disconnect_context; + struct PendingMessage *pending_pos; #if DEBUG_DV GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "%s: Receives core peer disconnect message!\n", "dv"); #endif + /* Check for disconnect from self message */ + if (0 == memcmp(&my_identity, peer, sizeof(struct GNUNET_PeerIdentity))) + return; + neighbor = - GNUNET_CONTAINER_multihashmap_get (ctx.direct_neighbors, &peer->hashPubKey); + GNUNET_CONTAINER_multihashmap_get (direct_neighbors, &peer->hashPubKey); + if (neighbor == NULL) { return; } + + pending_pos = core_pending_head; + while (NULL != pending_pos) + { + if (0 == memcmp(&pending_pos->recipient, &neighbor->identity, sizeof(struct GNUNET_PeerIdentity))) + { + GNUNET_CONTAINER_DLL_remove(core_pending_head, core_pending_tail, pending_pos); + pending_pos = core_pending_head; + } + else + pending_pos = pending_pos->next; + } + while (NULL != (referee = neighbor->referee_head)) distant_neighbor_free (referee); fdc.dest = NULL; fdc.tid = 0; - GNUNET_CONTAINER_multihashmap_iterate (ctx.extended_neighbors, &find_distant_peer, &fdc); + GNUNET_CONTAINER_multihashmap_iterate (extended_neighbors, &find_distant_peer, &fdc); if (fdc.dest != NULL) { disconnect_context.direct = neighbor; disconnect_context.distant = fdc.dest; - GNUNET_CONTAINER_multihashmap_iterate (ctx.direct_neighbors, &schedule_disconnect_messages, &disconnect_context); + GNUNET_CONTAINER_multihashmap_iterate (direct_neighbors, &schedule_disconnect_messages, &disconnect_context); } GNUNET_assert (neighbor->referee_tail == NULL); - if (GNUNET_NO == GNUNET_CONTAINER_multihashmap_remove (ctx.direct_neighbors, - &peer->hashPubKey, neighbor)) + if (GNUNET_NO == GNUNET_CONTAINER_multihashmap_remove (direct_neighbors, + &peer->hashPubKey, neighbor)) { GNUNET_break(0); } - if ((neighbor->send_context != NULL) && (neighbor->send_context->task != GNUNET_SCHEDULER_NO_TASK)) - GNUNET_SCHEDULER_cancel(sched, neighbor->send_context->task); + if ((neighbor->send_context != NULL) && + (neighbor->send_context->task != GNUNET_SCHEDULER_NO_TASK)) + GNUNET_SCHEDULER_cancel(neighbor->send_context->task); GNUNET_free (neighbor); } @@ -2810,53 +3042,50 @@ void handle_core_disconnect (void *cls, * Process dv 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) { unsigned long long max_hosts; - sched = scheduler; cfg = c; /* FIXME: Read from config, or calculate, or something other than this! */ max_hosts = DEFAULT_DIRECT_CONNECTIONS; - ctx.max_table_size = DEFAULT_DV_SIZE; - ctx.fisheye_depth = DEFAULT_FISHEYE_DEPTH; + max_table_size = DEFAULT_DV_SIZE; + fisheye_depth = DEFAULT_FISHEYE_DEPTH; if (GNUNET_CONFIGURATION_have_value(cfg, "dv", "max_direct_connections")) GNUNET_assert(GNUNET_OK == GNUNET_CONFIGURATION_get_value_number(cfg, "dv", "max_direct_connections", &max_hosts)); if (GNUNET_CONFIGURATION_have_value(cfg, "dv", "max_total_connections")) - GNUNET_assert(GNUNET_OK == GNUNET_CONFIGURATION_get_value_number(cfg, "dv", "max_total_connections", &ctx.max_table_size)); + GNUNET_assert(GNUNET_OK == GNUNET_CONFIGURATION_get_value_number(cfg, "dv", "max_total_connections", &max_table_size)); if (GNUNET_CONFIGURATION_have_value(cfg, "dv", "fisheye_depth")) - GNUNET_assert(GNUNET_OK == GNUNET_CONFIGURATION_get_value_number(cfg, "dv", "fisheye_depth", &ctx.fisheye_depth)); + GNUNET_assert(GNUNET_OK == GNUNET_CONFIGURATION_get_value_number(cfg, "dv", "fisheye_depth", &fisheye_depth)); - ctx.neighbor_min_heap = + neighbor_min_heap = GNUNET_CONTAINER_heap_create (GNUNET_CONTAINER_HEAP_ORDER_MIN); - ctx.neighbor_max_heap = + neighbor_max_heap = GNUNET_CONTAINER_heap_create (GNUNET_CONTAINER_HEAP_ORDER_MAX); - ctx.direct_neighbors = GNUNET_CONTAINER_multihashmap_create (max_hosts); - ctx.extended_neighbors = - GNUNET_CONTAINER_multihashmap_create (ctx.max_table_size * 3); + direct_neighbors = GNUNET_CONTAINER_multihashmap_create (max_hosts); + extended_neighbors = + GNUNET_CONTAINER_multihashmap_create (max_table_size * 3); GNUNET_SERVER_add_handlers (server, plugin_handlers); coreAPI = - GNUNET_CORE_connect (sched, - cfg, - GNUNET_TIME_relative_get_forever(), + GNUNET_CORE_connect (cfg, + 1, NULL, /* FIXME: anything we want to pass around? */ &core_init, &handle_core_connect, &handle_core_disconnect, + NULL, NULL, GNUNET_NO, NULL, @@ -2866,7 +3095,10 @@ run (void *cls, if (coreAPI == NULL) return; - peerinfo_handle = GNUNET_PEERINFO_connect(sched, cfg); + coreMST = GNUNET_SERVER_mst_create (&tokenized_message_handler, + NULL); + + peerinfo_handle = GNUNET_PEERINFO_connect(cfg); if (peerinfo_handle == NULL) { @@ -2875,16 +3107,9 @@ run (void *cls, } /* 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); -#if INSANE_GOSSIP - GNUNET_SCHEDULER_add_delayed (sched, - GNUNET_TIME_relative_multiply(GNUNET_TIME_UNIT_SECONDS, 5), - &gossip_all_to_all, - NULL); -#endif + cleanup_task = GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_FOREVER_REL, + &shutdown_task, + NULL); }