X-Git-Url: https://git.librecmc.org/?a=blobdiff_plain;f=src%2Fnse%2Fgnunet-service-nse.c;h=d7fb439e2269abccbf2678925ea196a2f6aca488;hb=a60b958f984d08525b636a2c7eae564ebec54ae6;hp=b0d635be6a7de51b855b0551ea17ecf7a635899a;hpb=61a374c33cac36eea8a4c1c5e85064b3a78608e2;p=oweals%2Fgnunet.git diff --git a/src/nse/gnunet-service-nse.c b/src/nse/gnunet-service-nse.c index b0d635be6..d7fb439e2 100644 --- a/src/nse/gnunet-service-nse.c +++ b/src/nse/gnunet-service-nse.c @@ -1,6 +1,6 @@ /* This file is part of GNUnet. - (C) 2009, 2010, 2011 Christian Grothoff (and other contributing authors) + (C) 2009, 2010, 2011, 2012 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 @@ -64,18 +64,17 @@ /** * Over how many values do we calculate the weighted average? */ -#define HISTORY_SIZE 8 - -/** - * Size of the queue to core. - */ -#define CORE_QUEUE_SIZE 2 +#define HISTORY_SIZE 64 /** * Message priority to use. */ #define NSE_PRIORITY 5 +#if FREEBSD +#define log2(a) (log(a)/log(2)) +#endif + /** * Amount of work required (W-bit collisions) for NSE proofs, in collision-bits. */ @@ -105,11 +104,6 @@ static struct GNUNET_BIO_WriteHandle *wh; struct NSEPeerEntry { - /** - * Pending message for this peer. - */ - struct GNUNET_MessageHeader *pending_message; - /** * Core handle for sending messages to this peer. */ @@ -131,9 +125,31 @@ struct NSEPeerEntry * been taken care of. */ int previous_round; + +#if ENABLE_HISTOGRAM + + /** + * Amount of messages received from this peer on this round. + */ + unsigned int received_messages; + + /** + * Amount of messages transmitted to this peer on this round. + */ + unsigned int transmitted_messages; + + /** + * Which size did we tell the peer the network is? + */ + unsigned int last_transmitted_size; + +#endif + }; +GNUNET_NETWORK_STRUCT_BEGIN + /** * Network size estimate reply; sent when "this" * peer's timer has run out before receiving a @@ -184,7 +200,7 @@ struct GNUNET_NSE_FloodMessage */ struct GNUNET_CRYPTO_RsaSignature signature; }; - +GNUNET_NETWORK_STRUCT_END /** * Handle to our current configuration. @@ -288,6 +304,16 @@ static struct GNUNET_PeerIdentity my_identity; */ static uint64_t my_proof; +/** + * Handle to this serivce's server. + */ +static struct GNUNET_SERVER_Handle *srv; + +/** + * Hostkey generation context + */ +static struct GNUNET_CRYPTO_RsaKeyGenerationContext *keygen; + /** * Initialize a message to clients with the current network @@ -299,52 +325,80 @@ static void setup_estimate_message (struct GNUNET_NSE_ClientMessage *em) { unsigned int i; + unsigned int j; double mean; double sum; double std_dev; double variance; double val; - double weight; + double nsize; + +#define WEST 1 + /* Weighted incremental algorithm for stddev according to West (1979) */ +#if WEST double sumweight; + double weight; double q; double r; double temp; - double nsize; - /* Weighted incremental algorithm for stddev according to West (1979) */ mean = 0.0; sum = 0.0; sumweight = 0.0; + variance = 0.0; for (i = 0; i < estimate_count; i++) { - val = - htonl (size_estimate_messages - [(estimate_index - i + - HISTORY_SIZE) % HISTORY_SIZE].matching_bits); - weight = 1; /* was: estimate_count + 1 - i; */ + j = (estimate_index - i + HISTORY_SIZE) % HISTORY_SIZE; + val = htonl (size_estimate_messages[j].matching_bits); + weight = estimate_count + 1 - i; temp = weight + sumweight; q = val - mean; r = q * weight / temp; - sum += sumweight * q * r; mean += r; + sum += sumweight * q * r; sumweight = temp; } - variance = sum / (sumweight - 1.0); - GNUNET_assert (variance >= 0); - std_dev = sqrt (variance); + if (estimate_count > 0) + variance = (sum / sumweight) * estimate_count / (estimate_count - 1.0); +#else + /* trivial version for debugging */ + double vsq; + + /* non-weighted trivial version */ + sum = 0.0; + vsq = 0.0; + variance = 0.0; + mean = 0.0; + + for (i = 0; i < estimate_count; i++) + { + j = (estimate_index - i + HISTORY_SIZE) % HISTORY_SIZE; + val = htonl (size_estimate_messages[j].matching_bits); + sum += val; + vsq += val * val; + } + if (0 != estimate_count) + { + mean = sum / estimate_count; + variance = (vsq - mean * sum) / (estimate_count - 1.0); // terrible for numerical stability... + } +#endif + if (variance >= 0) + std_dev = sqrt (variance); + else + std_dev = variance; /* must be infinity due to estimate_count == 0 */ current_std_dev = std_dev; current_size_estimate = mean; - + em->header.size = htons (sizeof (struct GNUNET_NSE_ClientMessage)); em->header.type = htons (GNUNET_MESSAGE_TYPE_NSE_ESTIMATE); em->reserved = htonl (0); em->timestamp = GNUNET_TIME_absolute_hton (GNUNET_TIME_absolute_get ()); - em->size_estimate = mean - 0.332747; + double se = mean - 0.332747; nsize = log2 (GNUNET_CONTAINER_multihashmap_size (peers) + 1); - if (em->size_estimate < nsize) - em->size_estimate = nsize; - em->std_deviation = std_dev; + em->size_estimate = GNUNET_hton_double (GNUNET_MAX (se, nsize)); + em->std_deviation = GNUNET_hton_double (std_dev); GNUNET_STATISTICS_set (stats, "# nodes in the network (estimate)", (uint64_t) pow (2, mean - 1.0 / 3.0), GNUNET_NO); } @@ -366,9 +420,7 @@ handle_start_message (void *cls, struct GNUNET_SERVER_Client *client, { struct GNUNET_NSE_ClientMessage em; -#if DEBUG_NSE GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Received START message from client\n"); -#endif GNUNET_SERVER_notification_context_add (nc, client); setup_estimate_message (&em); GNUNET_SERVER_notification_context_unicast (nc, client, &em.header, @@ -408,15 +460,15 @@ get_delay_randomization (uint32_t matching_bits) { #if USE_RANDOM_DELAYS struct GNUNET_TIME_Relative ret; + uint32_t i; + double d; - if (matching_bits == 0) - return GNUNET_TIME_UNIT_ZERO; - ret.rel_value = - GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK, - (uint32_t) (get_matching_bits_delay - (matching_bits - - 1) / (double) (hop_count_max + - 1))); + d = get_matching_bits_delay (matching_bits); + i = (uint32_t) (d / (double) (hop_count_max + 1)); + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, + "Randomizing flood using latencies up to %u ms\n", + (unsigned int) i); + ret.rel_value = GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK, i + 1); return ret; #else return GNUNET_TIME_UNIT_ZERO; @@ -435,7 +487,7 @@ static uint32_t get_matching_bits (struct GNUNET_TIME_Absolute timestamp, const struct GNUNET_PeerIdentity *id) { - GNUNET_HashCode timestamp_hash; + struct GNUNET_HashCode timestamp_hash; GNUNET_CRYPTO_hash (×tamp.abs_value, sizeof (timestamp.abs_value), ×tamp_hash); @@ -468,11 +520,9 @@ get_transmit_delay (int round_offset) #else ret = GNUNET_TIME_UNIT_ZERO; #endif -#if DEBUG_NSE GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Transmitting previous round behind schedule in %llu ms\n", (unsigned long long) ret.rel_value); -#endif return ret; case 0: /* current round is based on best-known matching_bits */ @@ -481,13 +531,11 @@ get_transmit_delay (int round_offset) dist_delay = get_matching_bits_delay (matching_bits); dist_delay += get_delay_randomization (matching_bits).rel_value; ret.rel_value = (uint64_t) dist_delay; -#if DEBUG_NSE GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "For round %llu, delay for %u matching bits is %llu ms\n", (unsigned long long) current_timestamp.abs_value, (unsigned int) matching_bits, (unsigned long long) ret.rel_value); -#endif /* now consider round start time and add delay to it */ tgt = GNUNET_TIME_absolute_add (current_timestamp, ret); return GNUNET_TIME_absolute_get_remaining (tgt); @@ -504,7 +552,7 @@ get_transmit_delay (int round_offset) * @param tc scheduler context */ static void -transmit_task (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc); +transmit_task_cb (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc); /** @@ -523,22 +571,22 @@ transmit_ready (void *cls, size_t size, void *buf) unsigned int idx; peer_entry->th = NULL; - if (buf == NULL) + if (NULL == buf) { /* client disconnected */ return 0; } GNUNET_assert (size >= sizeof (struct GNUNET_NSE_FloodMessage)); idx = estimate_index; - if (peer_entry->previous_round == GNUNET_NO) + if (GNUNET_NO == peer_entry->previous_round) { idx = (idx + HISTORY_SIZE - 1) % HISTORY_SIZE; peer_entry->previous_round = GNUNET_YES; peer_entry->transmit_task = - GNUNET_SCHEDULER_add_delayed (get_transmit_delay (0), &transmit_task, + GNUNET_SCHEDULER_add_delayed (get_transmit_delay (0), &transmit_task_cb, peer_entry); } - if ((ntohl (size_estimate_messages[idx].hop_count) == 0) && + if ((0 == ntohl (size_estimate_messages[idx].hop_count)) && (GNUNET_SCHEDULER_NO_TASK != proof_task)) { GNUNET_STATISTICS_update (stats, @@ -546,7 +594,13 @@ transmit_ready (void *cls, size_t size, void *buf) 1, GNUNET_NO); return 0; } -#if DEBUG_NSE + if (0 == ntohs (size_estimate_messages[idx].header.size)) + { + GNUNET_STATISTICS_update (stats, + "# flood messages not generated (lack of history)", + 1, GNUNET_NO); + return 0; + } GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "In round %llu, sending to `%s' estimate with %u bits\n", (unsigned long long) @@ -554,14 +608,17 @@ transmit_ready (void *cls, size_t size, void *buf) timestamp).abs_value, GNUNET_i2s (&peer_entry->id), (unsigned int) ntohl (size_estimate_messages[idx].matching_bits)); -#endif if (ntohl (size_estimate_messages[idx].hop_count) == 0) GNUNET_STATISTICS_update (stats, "# flood messages started", 1, GNUNET_NO); GNUNET_STATISTICS_update (stats, "# flood messages transmitted", 1, GNUNET_NO); +#if ENABLE_HISTOGRAM + peer_entry->transmitted_messages++; + peer_entry->last_transmitted_size = + ntohl(size_estimate_messages[idx].matching_bits); +#endif memcpy (buf, &size_estimate_messages[idx], sizeof (struct GNUNET_NSE_FloodMessage)); - GNUNET_STATISTICS_update (stats, "# flood messages sent", 1, GNUNET_NO); return sizeof (struct GNUNET_NSE_FloodMessage); } @@ -573,11 +630,12 @@ transmit_ready (void *cls, size_t size, void *buf) * @param tc scheduler context */ static void -transmit_task (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc) +transmit_task_cb (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc) { struct NSEPeerEntry *peer_entry = cls; peer_entry->transmit_task = GNUNET_SCHEDULER_NO_TASK; + GNUNET_assert (NULL == peer_entry->th); peer_entry->th = GNUNET_CORE_notify_transmit_ready (coreAPI, GNUNET_NO, NSE_PRIORITY, @@ -613,7 +671,8 @@ update_network_size_estimate () * @param ts timestamp to use */ static void -setup_flood_message (unsigned int slot, struct GNUNET_TIME_Absolute ts) +setup_flood_message (unsigned int slot, + struct GNUNET_TIME_Absolute ts) { struct GNUNET_NSE_FloodMessage *fm; uint32_t matching_bits; @@ -632,8 +691,12 @@ setup_flood_message (unsigned int slot, struct GNUNET_TIME_Absolute ts) fm->timestamp = GNUNET_TIME_absolute_hton (ts); fm->pkey = my_public_key; fm->proof_of_work = my_proof; - GNUNET_assert (GNUNET_OK == - GNUNET_CRYPTO_rsa_sign (my_private_key, &fm->purpose, &fm->signature)); + if (nse_work_required > 0) + GNUNET_assert (GNUNET_OK == + GNUNET_CRYPTO_rsa_sign (my_private_key, &fm->purpose, + &fm->signature)); + else + memset (&fm->signature, 0, sizeof (fm->signature)); } @@ -647,25 +710,35 @@ setup_flood_message (unsigned int slot, struct GNUNET_TIME_Absolute ts) * @return GNUNET_OK (continue to iterate) */ static int -schedule_current_round (void *cls, const GNUNET_HashCode * key, void *value) +schedule_current_round (void *cls, + const struct GNUNET_HashCode * key, + void *value) { struct NSEPeerEntry *peer_entry = value; struct GNUNET_TIME_Relative delay; - if (peer_entry->th != NULL) + if (NULL != peer_entry->th) { peer_entry->previous_round = GNUNET_NO; return GNUNET_OK; } - if (peer_entry->transmit_task != GNUNET_SCHEDULER_NO_TASK) + if (GNUNET_SCHEDULER_NO_TASK != peer_entry->transmit_task) { GNUNET_SCHEDULER_cancel (peer_entry->transmit_task); peer_entry->previous_round = GNUNET_NO; } +#if ENABLE_HISTOGRAM + if (peer_entry->received_messages > 1) + GNUNET_STATISTICS_update(stats, "# extra messages", + peer_entry->received_messages - 1, GNUNET_NO); + peer_entry->transmitted_messages = 0; + peer_entry->last_transmitted_size = 0; + peer_entry->received_messages = 0; +#endif delay = get_transmit_delay ((peer_entry->previous_round == GNUNET_NO) ? -1 : 0); peer_entry->transmit_task = - GNUNET_SCHEDULER_add_delayed (delay, &transmit_task, peer_entry); + GNUNET_SCHEDULER_add_delayed (delay, &transmit_task_cb, peer_entry); return GNUNET_OK; } @@ -683,6 +756,8 @@ update_flood_message (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc) unsigned int i; flood_task = GNUNET_SCHEDULER_NO_TASK; + if (0 != (tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN)) + return; offset = GNUNET_TIME_absolute_get_remaining (next_timestamp); if (0 != offset.rel_value) { @@ -691,14 +766,16 @@ update_flood_message (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc) GNUNET_SCHEDULER_add_delayed (offset, &update_flood_message, NULL); return; } - current_timestamp = next_timestamp; - next_timestamp = - GNUNET_TIME_absolute_add (current_timestamp, gnunet_nse_interval); estimate_index = (estimate_index + 1) % HISTORY_SIZE; if (estimate_count < HISTORY_SIZE) estimate_count++; - if (next_timestamp.abs_value == - GNUNET_TIME_absolute_ntoh (next_message.timestamp).abs_value) + current_timestamp = next_timestamp; + next_timestamp = + GNUNET_TIME_absolute_add (current_timestamp, gnunet_nse_interval); + if ((current_timestamp.abs_value == + GNUNET_TIME_absolute_ntoh (next_message.timestamp).abs_value) && + (get_matching_bits (current_timestamp, &my_identity) < + ntohl(next_message.matching_bits))) { /* we received a message for this round way early, use it! */ size_estimate_messages[estimate_index] = next_message; @@ -727,7 +804,7 @@ update_flood_message (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc) * @return the number of leading zero bits. */ static unsigned int -count_leading_zeroes (const GNUNET_HashCode * hash) +count_leading_zeroes (const struct GNUNET_HashCode * hash) { unsigned int hash_count; @@ -752,8 +829,8 @@ check_proof_of_work (const struct GNUNET_CRYPTO_RsaPublicKeyBinaryEncoded *pkey, uint64_t val) { char buf[sizeof (struct GNUNET_CRYPTO_RsaPublicKeyBinaryEncoded) + - sizeof (val)]; - GNUNET_HashCode result; + sizeof (val)] GNUNET_ALIGN; + struct GNUNET_HashCode result; memcpy (buf, &val, sizeof (val)); memcpy (&buf[sizeof (val)], pkey, @@ -797,8 +874,8 @@ find_proof (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc) #define ROUND_SIZE 10 uint64_t counter; char buf[sizeof (struct GNUNET_CRYPTO_RsaPublicKeyBinaryEncoded) + - sizeof (uint64_t)]; - GNUNET_HashCode result; + sizeof (uint64_t)] GNUNET_ALIGN; + struct GNUNET_HashCode result; unsigned int i; proof_task = GNUNET_SCHEDULER_NO_TASK; @@ -813,20 +890,10 @@ find_proof (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc) if (nse_work_required <= count_leading_zeroes (&result)) { my_proof = counter; -#if DEBUG_NSE GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Proof of work found: %llu!\n", (unsigned long long) GNUNET_ntohll (counter)); -#endif - for (i = 0; i < HISTORY_SIZE; i++) - if (ntohl (size_estimate_messages[i].hop_count) == 0) - { - size_estimate_messages[i].proof_of_work = my_proof; - GNUNET_assert (GNUNET_OK == - GNUNET_CRYPTO_rsa_sign (my_private_key, - &size_estimate_messages[i].purpose, - &size_estimate_messages[i].signature)); - } write_proof (); + setup_flood_message (estimate_index, current_timestamp); return; } counter++; @@ -834,10 +901,8 @@ find_proof (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc) } if (my_proof / (100 * ROUND_SIZE) < counter / (100 * ROUND_SIZE)) { -#if DEBUG_NSE GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Testing proofs currently at %llu\n", (unsigned long long) counter); -#endif /* remember progress every 100 rounds */ my_proof = counter; write_proof (); @@ -847,7 +912,9 @@ find_proof (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc) my_proof = counter; } proof_task = - GNUNET_SCHEDULER_add_delayed (proof_find_delay, &find_proof, NULL); + GNUNET_SCHEDULER_add_delayed_with_priority (proof_find_delay, + GNUNET_SCHEDULER_PRIORITY_IDLE, + &find_proof, NULL); } @@ -874,11 +941,12 @@ verify_message_crypto (const struct GNUNET_NSE_FloodMessage *incoming_flood) GNUNET_break_op (0); return GNUNET_NO; } - if (GNUNET_OK != - GNUNET_CRYPTO_rsa_verify (GNUNET_SIGNATURE_PURPOSE_NSE_SEND, - &incoming_flood->purpose, - &incoming_flood->signature, - &incoming_flood->pkey)) + if ((nse_work_required > 0) && + (GNUNET_OK != + GNUNET_CRYPTO_rsa_verify (GNUNET_SIGNATURE_PURPOSE_NSE_SEND, + &incoming_flood->purpose, + &incoming_flood->signature, + &incoming_flood->pkey))) { GNUNET_break_op (0); return GNUNET_NO; @@ -897,13 +965,13 @@ verify_message_crypto (const struct GNUNET_NSE_FloodMessage *incoming_flood) * @return GNUNET_OK (continue to iterate) */ static int -update_flood_times (void *cls, const GNUNET_HashCode * key, void *value) +update_flood_times (void *cls, const struct GNUNET_HashCode * key, void *value) { struct NSEPeerEntry *exclude = cls; struct NSEPeerEntry *peer_entry = value; struct GNUNET_TIME_Relative delay; - if (peer_entry->th != NULL) + if (NULL != peer_entry->th) return GNUNET_OK; /* already active */ if (peer_entry == exclude) return GNUNET_OK; /* trigger of the update */ @@ -911,18 +979,21 @@ update_flood_times (void *cls, const GNUNET_HashCode * key, void *value) { /* still stuck in previous round, no point to update, check that * we are active here though... */ - GNUNET_break ((peer_entry->transmit_task != GNUNET_SCHEDULER_NO_TASK) || - (peer_entry->th != NULL)); + if ( (GNUNET_SCHEDULER_NO_TASK == peer_entry->transmit_task) && + (NULL == peer_entry->th) ) + { + GNUNET_break (0); + } return GNUNET_OK; } - if (peer_entry->transmit_task != GNUNET_SCHEDULER_NO_TASK) + if (GNUNET_SCHEDULER_NO_TASK != peer_entry->transmit_task) { GNUNET_SCHEDULER_cancel (peer_entry->transmit_task); peer_entry->transmit_task = GNUNET_SCHEDULER_NO_TASK; } delay = get_transmit_delay (0); peer_entry->transmit_task = - GNUNET_SCHEDULER_add_delayed (delay, &transmit_task, peer_entry); + GNUNET_SCHEDULER_add_delayed (delay, &transmit_task_cb, peer_entry); return GNUNET_OK; } @@ -934,12 +1005,13 @@ update_flood_times (void *cls, const GNUNET_HashCode * key, void *value) * @param message message * @param peer peer identity this message is from (ignored) * @param atsi performance data (ignored) - * + * @param atsi_count number of records in 'atsi' */ static int handle_p2p_size_estimate (void *cls, const struct GNUNET_PeerIdentity *peer, const struct GNUNET_MessageHeader *message, - const struct GNUNET_TRANSPORT_ATS_Information *atsi) + const struct GNUNET_ATS_Information *atsi, + unsigned int atsi_count) { const struct GNUNET_NSE_FloodMessage *incoming_flood; struct GNUNET_TIME_Absolute ts; @@ -949,7 +1021,7 @@ handle_p2p_size_estimate (void *cls, const struct GNUNET_PeerIdentity *peer, #if ENABLE_HISTOGRAM if (NULL != wh) - GNUNET_BIO_write_int64 (wh, GNUNET_TIME_absolute_get ().abs_value); + GNUNET_break (GNUNET_OK == GNUNET_BIO_write_int64 (wh, GNUNET_TIME_absolute_get ().abs_value)); #endif incoming_flood = (const struct GNUNET_NSE_FloodMessage *) message; GNUNET_STATISTICS_update (stats, "# flood messages received", 1, GNUNET_NO); @@ -980,16 +1052,20 @@ handle_p2p_size_estimate (void *cls, const struct GNUNET_PeerIdentity *peer, GNUNET_break (0); return GNUNET_OK; } +#if ENABLE_HISTOGRAM + peer_entry->received_messages++; + if (peer_entry->transmitted_messages > 0 && + peer_entry->last_transmitted_size >= matching_bits) + GNUNET_STATISTICS_update(stats, "# cross messages", 1, GNUNET_NO); +#endif ts = GNUNET_TIME_absolute_ntoh (incoming_flood->timestamp); - if (ts.abs_value == current_timestamp.abs_value) idx = estimate_index; else if (ts.abs_value == current_timestamp.abs_value - gnunet_nse_interval.rel_value) idx = (estimate_index + HISTORY_SIZE - 1) % HISTORY_SIZE; - else if (ts.abs_value == - next_timestamp.abs_value - gnunet_nse_interval.rel_value) + else if (ts.abs_value == next_timestamp.abs_value) { if (matching_bits <= ntohl (next_message.matching_bits)) return GNUNET_OK; /* ignore, simply too early/late */ @@ -1016,45 +1092,45 @@ handle_p2p_size_estimate (void *cls, const struct GNUNET_PeerIdentity *peer, update_network_size_estimate (); return GNUNET_OK; } - if (matching_bits >= ntohl (size_estimate_messages[idx].matching_bits)) + if (matching_bits == ntohl (size_estimate_messages[idx].matching_bits)) { - /* cancel transmission from us to this peer for this round */ - if (idx == estimate_index) + /* Cancel transmission in the other direction, as this peer clearly has + up-to-date information already. Even if we didn't talk to this peer in + the previous round, we should no longer send it stale information as it + told us about the current round! */ + peer_entry->previous_round = GNUNET_YES; + if (idx != estimate_index) { - if (peer_entry->previous_round == GNUNET_YES) - { - /* cancel any activity for current round */ - if (peer_entry->transmit_task != GNUNET_SCHEDULER_NO_TASK) - { - GNUNET_SCHEDULER_cancel (peer_entry->transmit_task); - peer_entry->transmit_task = GNUNET_SCHEDULER_NO_TASK; - } - if (peer_entry->th != NULL) - { - GNUNET_CORE_notify_transmit_ready_cancel (peer_entry->th); - peer_entry->th = NULL; - } - } + /* do not transmit information for the previous round to this peer + anymore (but allow current round) */ + return GNUNET_OK; } - else + /* got up-to-date information for current round, cancel transmission to + * this peer altogether */ + if (GNUNET_SCHEDULER_NO_TASK != peer_entry->transmit_task) { - /* cancel previous round only */ - peer_entry->previous_round = GNUNET_YES; + GNUNET_SCHEDULER_cancel (peer_entry->transmit_task); + peer_entry->transmit_task = GNUNET_SCHEDULER_NO_TASK; + } + if (NULL != peer_entry->th) + { + GNUNET_CORE_notify_transmit_ready_cancel (peer_entry->th); + peer_entry->th = NULL; } - } - if (matching_bits == ntohl (size_estimate_messages[idx].matching_bits)) return GNUNET_OK; - if (matching_bits <= ntohl (size_estimate_messages[idx].matching_bits)) + } + if (matching_bits < ntohl (size_estimate_messages[idx].matching_bits)) { - if ((idx < estimate_index) && (peer_entry->previous_round == GNUNET_YES)) + if ((idx < estimate_index) && (peer_entry->previous_round == GNUNET_YES)) { peer_entry->previous_round = GNUNET_NO; + } /* push back our result now, that peer is spreading bad information... */ if (NULL == peer_entry->th) { if (peer_entry->transmit_task != GNUNET_SCHEDULER_NO_TASK) GNUNET_SCHEDULER_cancel (peer_entry->transmit_task); peer_entry->transmit_task = - GNUNET_SCHEDULER_add_now (&transmit_task, peer_entry); + GNUNET_SCHEDULER_add_now (&transmit_task_cb, peer_entry); } /* Not closer than our most recent message, no need to do work here */ GNUNET_STATISTICS_update (stats, @@ -1067,11 +1143,34 @@ handle_p2p_size_estimate (void *cls, const struct GNUNET_PeerIdentity *peer, GNUNET_break_op (0); return GNUNET_OK; } + GNUNET_assert (matching_bits > + ntohl (size_estimate_messages[idx].matching_bits)); + /* Cancel transmission in the other direction, as this peer clearly has + * up-to-date information already. + */ + peer_entry->previous_round = GNUNET_YES; + if (idx == estimate_index) + { + /* cancel any activity for current round */ + if (peer_entry->transmit_task != GNUNET_SCHEDULER_NO_TASK) + { + GNUNET_SCHEDULER_cancel (peer_entry->transmit_task); + peer_entry->transmit_task = GNUNET_SCHEDULER_NO_TASK; + } + if (peer_entry->th != NULL) + { + GNUNET_CORE_notify_transmit_ready_cancel (peer_entry->th); + peer_entry->th = NULL; + } + } size_estimate_messages[idx] = *incoming_flood; size_estimate_messages[idx].hop_count = htonl (ntohl (incoming_flood->hop_count) + 1); hop_count_max = GNUNET_MAX (ntohl (incoming_flood->hop_count) + 1, hop_count_max); + GNUNET_STATISTICS_set (stats, + "# estimated network diameter", + hop_count_max, GNUNET_NO); /* have a new, better size estimate, inform clients */ update_network_size_estimate (); @@ -1085,34 +1184,39 @@ handle_p2p_size_estimate (void *cls, const struct GNUNET_PeerIdentity *peer, /** - * Method called whenever a peer connects. + * Method called whenever a peer connects. Sets up the PeerEntry and + * schedules the initial size info transmission to this peer. * * @param cls closure * @param peer peer identity this notification is about * @param atsi performance data + * @param atsi_count number of records in 'atsi' */ static void handle_core_connect (void *cls, const struct GNUNET_PeerIdentity *peer, - const struct GNUNET_TRANSPORT_ATS_Information *atsi) + const struct GNUNET_ATS_Information *atsi, + unsigned int atsi_count) { struct NSEPeerEntry *peer_entry; -#if DEBUG_NSE GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Peer `%s' connected to us\n", GNUNET_i2s (peer)); -#endif peer_entry = GNUNET_malloc (sizeof (struct NSEPeerEntry)); peer_entry->id = *peer; - GNUNET_CONTAINER_multihashmap_put (peers, &peer->hashPubKey, peer_entry, - GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY); + GNUNET_assert (GNUNET_OK == + GNUNET_CONTAINER_multihashmap_put (peers, &peer->hashPubKey, + peer_entry, + GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY)); peer_entry->transmit_task = - GNUNET_SCHEDULER_add_delayed (get_transmit_delay (-1), &transmit_task, + GNUNET_SCHEDULER_add_delayed (get_transmit_delay (-1), &transmit_task_cb, peer_entry); + GNUNET_STATISTICS_update (stats, "# peers connected", 1, GNUNET_NO); } /** - * Method called whenever a peer disconnects. + * Method called whenever a peer disconnects. Deletes the PeerEntry and cancels + * any pending transmission requests to that peer. * * @param cls closure * @param peer peer identity this notification is about @@ -1122,10 +1226,8 @@ handle_core_disconnect (void *cls, const struct GNUNET_PeerIdentity *peer) { struct NSEPeerEntry *pos; -#if DEBUG_NSE GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Peer `%s' disconnected from us\n", GNUNET_i2s (peer)); -#endif pos = GNUNET_CONTAINER_multihashmap_get (peers, &peer->hashPubKey); if (NULL == pos) { @@ -1135,14 +1237,17 @@ handle_core_disconnect (void *cls, const struct GNUNET_PeerIdentity *peer) GNUNET_assert (GNUNET_YES == GNUNET_CONTAINER_multihashmap_remove (peers, &peer->hashPubKey, pos)); - if (pos->transmit_task != GNUNET_SCHEDULER_NO_TASK) + if (pos->transmit_task != GNUNET_SCHEDULER_NO_TASK) { GNUNET_SCHEDULER_cancel (pos->transmit_task); - if (pos->th != NULL) + pos->transmit_task = GNUNET_SCHEDULER_NO_TASK; + } + if (NULL != pos->th) { GNUNET_CORE_notify_transmit_ready_cancel (pos->th); pos->th = NULL; } GNUNET_free (pos); + GNUNET_STATISTICS_update (stats, "# peers connected", -1, GNUNET_NO); } @@ -1155,46 +1260,51 @@ handle_core_disconnect (void *cls, const struct GNUNET_PeerIdentity *peer) static void shutdown_task (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc) { - if (flood_task != GNUNET_SCHEDULER_NO_TASK) + if (GNUNET_SCHEDULER_NO_TASK != flood_task) { GNUNET_SCHEDULER_cancel (flood_task); flood_task = GNUNET_SCHEDULER_NO_TASK; } - if (proof_task != GNUNET_SCHEDULER_NO_TASK) + if (GNUNET_SCHEDULER_NO_TASK != proof_task) { GNUNET_SCHEDULER_cancel (proof_task); proof_task = GNUNET_SCHEDULER_NO_TASK; write_proof (); /* remember progress */ } - if (nc != NULL) + if (NULL != keygen) + { + GNUNET_CRYPTO_rsa_key_create_stop (keygen); + keygen = NULL; + } + if (NULL != nc) { GNUNET_SERVER_notification_context_destroy (nc); nc = NULL; } - if (coreAPI != NULL) + if (NULL != coreAPI) { GNUNET_CORE_disconnect (coreAPI); coreAPI = NULL; } - if (stats != NULL) + if (NULL != stats) { GNUNET_STATISTICS_destroy (stats, GNUNET_NO); stats = NULL; } - if (peers != NULL) + if (NULL != peers) { GNUNET_CONTAINER_multihashmap_destroy (peers); peers = NULL; } - if (my_private_key != NULL) + if (NULL != my_private_key) { GNUNET_CRYPTO_rsa_key_free (my_private_key); my_private_key = NULL; } #if ENABLE_HISTOGRAM - if (wh != NULL) + if (NULL != wh) { - GNUNET_BIO_write_close (wh); + GNUNET_break (GNUNET_OK == GNUNET_BIO_write_close (wh)); wh = NULL; } #endif @@ -1207,22 +1317,17 @@ shutdown_task (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc) * @param cls service closure * @param server handle to the server for this service * @param identity the public identity of this peer - * @param publicKey the public key of this peer */ static void core_init (void *cls, struct GNUNET_CORE_Handle *server, - const struct GNUNET_PeerIdentity *identity, - const struct GNUNET_CRYPTO_RsaPublicKeyBinaryEncoded *publicKey) + const struct GNUNET_PeerIdentity *identity) { struct GNUNET_TIME_Absolute now; struct GNUNET_TIME_Absolute prev_time; - unsigned int i; - if (server == NULL) + if (NULL == server) { -#if DEBUG_NSE - GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Connection to core FAILED!\n"); -#endif + GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "Connection to core FAILED!\n"); GNUNET_SCHEDULER_shutdown (); return; } @@ -1233,18 +1338,19 @@ core_init (void *cls, struct GNUNET_CORE_Handle *server, current_timestamp.abs_value = (now.abs_value / gnunet_nse_interval.rel_value) * gnunet_nse_interval.rel_value; - next_timestamp.abs_value = - current_timestamp.abs_value + gnunet_nse_interval.rel_value; - - for (i = 0; i < HISTORY_SIZE; i++) + next_timestamp = + GNUNET_TIME_absolute_add (current_timestamp, gnunet_nse_interval); + estimate_index = HISTORY_SIZE - 1; + estimate_count = 0; + if (GNUNET_YES == check_proof_of_work (&my_public_key, my_proof)) { + int idx = (estimate_index + HISTORY_SIZE - 1) % HISTORY_SIZE; prev_time.abs_value = - current_timestamp.abs_value - (HISTORY_SIZE - i - - 1) * gnunet_nse_interval.rel_value; - setup_flood_message (i, prev_time); + current_timestamp.abs_value - gnunet_nse_interval.rel_value; + setup_flood_message (idx, prev_time); + setup_flood_message (estimate_index, current_timestamp); + estimate_count++; } - estimate_index = HISTORY_SIZE - 1; - estimate_count = 2; flood_task = GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_absolute_get_remaining (next_timestamp), &update_flood_message, @@ -1253,19 +1359,17 @@ core_init (void *cls, struct GNUNET_CORE_Handle *server, /** - * Handle network size estimate clients. + * Callback for hostkey read/generation * - * @param cls closure - * @param server the initialized server - * @param c configuration to use + * @param cls NULL + * @param pk the private key + * @param emsg error message */ static void -run (void *cls, struct GNUNET_SERVER_Handle *server, - const struct GNUNET_CONFIGURATION_Handle *c) +key_generation_cb (void *cls, + struct GNUNET_CRYPTO_RsaPrivateKey *pk, + const char *emsg) { - char *keyfile; - char *proof; - static const struct GNUNET_SERVER_MessageHandler handlers[] = { {&handle_start_message, NULL, GNUNET_MESSAGE_TYPE_NSE_START, sizeof (struct GNUNET_MessageHeader)}, @@ -1276,52 +1380,18 @@ run (void *cls, struct GNUNET_SERVER_Handle *server, sizeof (struct GNUNET_NSE_FloodMessage)}, {NULL, 0, 0} }; - cfg = c; - - if ((GNUNET_OK != - GNUNET_CONFIGURATION_get_value_time (cfg, "NSE", "INTERVAL", - &gnunet_nse_interval)) || - (GNUNET_OK != - GNUNET_CONFIGURATION_get_value_time (cfg, "NSE", "WORKDELAY", - &proof_find_delay)) || - (GNUNET_OK != - GNUNET_CONFIGURATION_get_value_number (cfg, "NSE", "WORKBITS", - &nse_work_required))) - { - GNUNET_log (GNUNET_ERROR_TYPE_ERROR, - _ - ("NSE service is lacking key configuration settings. Exiting.\n")); - GNUNET_SCHEDULER_shutdown (); - return; - } - if (nse_work_required >= sizeof (GNUNET_HashCode) * 8) - { - GNUNET_log (GNUNET_ERROR_TYPE_ERROR, - _("Invalid work requirement for NSE service. Exiting.\n")); - GNUNET_SCHEDULER_shutdown (); - return; - } - + char *proof; - if (GNUNET_OK != - GNUNET_CONFIGURATION_get_value_filename (cfg, "GNUNETD", "HOSTKEY", - &keyfile)) - { - GNUNET_log (GNUNET_ERROR_TYPE_ERROR, - _ - ("NSE service is lacking key configuration settings. Exiting.\n")); - GNUNET_SCHEDULER_shutdown (); - return; - } - my_private_key = GNUNET_CRYPTO_rsa_key_create_from_file (keyfile); - GNUNET_free (keyfile); - if (my_private_key == NULL) + keygen = NULL; + if (NULL == pk) { GNUNET_log (GNUNET_ERROR_TYPE_ERROR, - _("NSE service could not access hostkey. Exiting.\n")); + _("NSE service could not access hostkey: %s\n"), + emsg); GNUNET_SCHEDULER_shutdown (); return; } + my_private_key = pk; GNUNET_CRYPTO_rsa_key_get_public (my_private_key, &my_public_key); GNUNET_CRYPTO_hash (&my_public_key, sizeof (my_public_key), &my_identity.hashPubKey); @@ -1331,11 +1401,8 @@ run (void *cls, struct GNUNET_SERVER_Handle *server, GNUNET_log (GNUNET_ERROR_TYPE_ERROR, _ ("NSE service is lacking key configuration settings. Exiting.\n")); - if (my_private_key != NULL) - { - GNUNET_CRYPTO_rsa_key_free (my_private_key); - my_private_key = NULL; - } + GNUNET_CRYPTO_rsa_key_free (my_private_key); + my_private_key = NULL; GNUNET_SCHEDULER_shutdown (); return; } @@ -1348,24 +1415,25 @@ run (void *cls, struct GNUNET_SERVER_Handle *server, GNUNET_SCHEDULER_add_with_priority (GNUNET_SCHEDULER_PRIORITY_IDLE, &find_proof, NULL); - peers = GNUNET_CONTAINER_multihashmap_create (128); - GNUNET_SERVER_add_handlers (server, handlers); - nc = GNUNET_SERVER_notification_context_create (server, 1); + peers = GNUNET_CONTAINER_multihashmap_create (128, GNUNET_NO); + GNUNET_SERVER_add_handlers (srv, handlers); + nc = GNUNET_SERVER_notification_context_create (srv, 1); /* Connect to core service and register core handlers */ coreAPI = GNUNET_CORE_connect (cfg, /* Main configuration */ - CORE_QUEUE_SIZE, /* queue size */ - NULL, /* Closure passed to functions */ + NULL, /* Closure passed to functions */ &core_init, /* Call core_init once connected */ &handle_core_connect, /* Handle connects */ &handle_core_disconnect, /* Handle disconnects */ - NULL, /* Do we care about "status" updates? */ NULL, /* Don't want notified about all incoming messages */ GNUNET_NO, /* For header only inbound notification */ NULL, /* Don't want notified about all outbound messages */ GNUNET_NO, /* For header only outbound notification */ core_handlers); /* Register these handlers */ - GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_FOREVER_REL, &shutdown_task, - NULL); + if (NULL == coreAPI) + { + GNUNET_SCHEDULER_shutdown (); + return; + } #if ENABLE_HISTOGRAM if (GNUNET_OK == GNUNET_CONFIGURATION_get_value_filename (cfg, "NSE", "HISTOGRAM", &proof)) @@ -1374,17 +1442,70 @@ run (void *cls, struct GNUNET_SERVER_Handle *server, GNUNET_free (proof); } #endif - if (coreAPI == NULL) + stats = GNUNET_STATISTICS_create ("nse", cfg); + GNUNET_SERVER_resume (srv); +} + + +/** + * Handle network size estimate clients. + * + * @param cls closure + * @param server the initialized server + * @param c configuration to use + */ +static void +run (void *cls, + struct GNUNET_SERVER_Handle *server, + const struct GNUNET_CONFIGURATION_Handle *c) +{ + char *keyfile; + + cfg = c; + srv = server; + if ((GNUNET_OK != + GNUNET_CONFIGURATION_get_value_time (cfg, "NSE", "INTERVAL", + &gnunet_nse_interval)) || + (GNUNET_OK != + GNUNET_CONFIGURATION_get_value_time (cfg, "NSE", "WORKDELAY", + &proof_find_delay)) || + (GNUNET_OK != + GNUNET_CONFIGURATION_get_value_number (cfg, "NSE", "WORKBITS", + &nse_work_required))) { + GNUNET_log (GNUNET_ERROR_TYPE_ERROR, + _ + ("NSE service is lacking key configuration settings. Exiting.\n")); GNUNET_SCHEDULER_shutdown (); return; } - stats = GNUNET_STATISTICS_create ("nse", cfg); + if (nse_work_required >= sizeof (struct GNUNET_HashCode) * 8) + { + GNUNET_log (GNUNET_ERROR_TYPE_ERROR, + _("Invalid work requirement for NSE service. Exiting.\n")); + GNUNET_SCHEDULER_shutdown (); + return; + } + if (GNUNET_OK != + GNUNET_CONFIGURATION_get_value_filename (cfg, "GNUNETD", "HOSTKEY", + &keyfile)) + { + GNUNET_log (GNUNET_ERROR_TYPE_ERROR, + _ + ("NSE service is lacking key configuration settings. Exiting.\n")); + GNUNET_SCHEDULER_shutdown (); + return; + } + GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_FOREVER_REL, &shutdown_task, + NULL); + GNUNET_SERVER_suspend (srv); + keygen = GNUNET_CRYPTO_rsa_key_create_start (keyfile, &key_generation_cb, NULL); + GNUNET_free (keyfile); } /** - * The main function for the statistics service. + * The main function for the network size estimation service. * * @param argc number of arguments from the command line * @param argv command line arguments