src/nse/gnunet-service-nse.c

   1 /*
   2  This file is part of GNUnet.
   3  (C) 2009, 2010, 2011 Christian Grothoff (and other contributing authors)
   4
   5  GNUnet is free software; you can redistribute it and/or modify
   6  it under the terms of the GNU General Public License as published
   7  by the Free Software Foundation; either version 3, or (at your
   8  option) any later version.
   9
  10  GNUnet is distributed in the hope that it will be useful, but
  11  WITHOUT ANY WARRANTY; without even the implied warranty of
  12  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13  General Public License for more details.
  14
  15  You should have received a copy of the GNU General Public License
  16  along with GNUnet; see the file COPYING.  If not, write to the
  17  Free Software Foundation, Inc., 59 Temple Place - Suite 330,
  18  Boston, MA 02111-1307, USA.
  19  */
  20
  21 /**
  22  * @file nse/gnunet-service-nse.c
  23  * @brief network size estimation service
  24  * @author Nathan Evans
  25  *
  26  * The purpose of this service is to estimate the size of the network.
  27  * Given a specified interval, each peer hashes the most recent
  28  * timestamp which is evenly divisible by that interval.  This hash
  29  * is compared in distance to the peer identity to choose an offset.
  30  * The closer the peer identity to the hashed timestamp, the earlier
  31  * the peer sends out a "nearest peer" message.  The closest peer's
  32  * message should thus be received before any others, which stops
  33  * those peer from sending their messages at a later duration.  So
  34  * every peer should receive the same nearest peer message, and
  35  * from this can calculate the expected number of peers in the
  36  * network.
  37  *
  38  */
  39 #include "platform.h"
  40 #include "gnunet_client_lib.h"
  41 #include "gnunet_constants.h"
  42 #include "gnunet_container_lib.h"
  43 #include "gnunet_protocols.h"
  44 #include "gnunet_signatures.h"
  45 #include "gnunet_service_lib.h"
  46 #include "gnunet_server_lib.h"
  47 #include "gnunet_statistics_service.h"
  48 #include "gnunet_core_service.h"
  49 #include "gnunet_time_lib.h"
  50 #include "gnunet_nse_service.h"
  51 #include "nse.h"
  52
  53 #define DEFAULT_HISTORY_SIZE 50
  54
  55 #define DEFAULT_CORE_QUEUE_SIZE 32
  56
  57 #define DEFAULT_NSE_PRIORITY 5
  58
  59 /**
  60  * Entry in the list of clients which
  61  * should be notified upon a new network
  62  * size estimate calculation.
  63  */
  64 struct ClientListEntry
  65 {
  66   /**
  67    *  Pointer to previous entry
  68    */
  69   struct ClientListEntry *prev;
  70
  71   /**
  72    *  Pointer to next entry
  73    */
  74   struct ClientListEntry *next;
  75
  76   /**
  77    * Client to notify.
  78    */
  79   struct GNUNET_SERVER_Client *client;
  80 };
  81
  82 /**
  83  * Per-peer information.
  84  */
  85 struct NSEPeerEntry
  86 {
  87   /**
  88    * Next peer entry (DLL)
  89    */
  90   struct NSEPeerEntry *next;
  91
  92   /**
  93    *  Prev peer entry (DLL)
  94    */
  95   struct NSEPeerEntry *prev;
  96
  97   /**
  98    * Pending message for this peer.
  99    */
 100   struct GNUNET_MessageHeader *pending_message;
 101
 102   /**
 103    * Core handle for sending messages to this peer.
 104    */
 105   struct GNUNET_CORE_TransmitHandle *th;
 106
 107   /**
 108    * What is the identity of the peer?
 109    */
 110   struct GNUNET_PeerIdentity id;
 111 };
 112
 113 /**
 114  * Handle to our current configuration.
 115  */
 116 static const struct GNUNET_CONFIGURATION_Handle *cfg;
 117
 118 /**
 119  * Handle to the statistics service.
 120  */
 121 static struct GNUNET_STATISTICS_Handle *stats;
 122
 123 /**
 124  * Handle to the core service.
 125  */
 126 static struct GNUNET_CORE_Handle *coreAPI;
 127
 128 /**
 129  * Head of global list of peers.
 130  */
 131 static struct NSEPeerEntry *peers_head;
 132
 133 /**
 134  * Head of global list of clients.
 135  */
 136 static struct NSEPeerEntry *peers_tail;
 137
 138 /**
 139  * Head of global list of clients.
 140  */
 141 static struct ClientListEntry *cle_head;
 142
 143 /**
 144  * Tail of global list of clients.
 145  */
 146 static struct ClientListEntry *cle_tail;
 147
 148 /**
 149  * The current network size estimate.
 150  * Number of bits matching on average
 151  * thus far.
 152  */
 153 static double current_size_estimate;
 154
 155 /**
 156  * The standard deviation of the last
 157  * DEFAULT_HISTORY_SIZE network size estimates.
 158  */
 159 static double current_std_dev;
 160
 161 /**
 162  * Array of the last DEFAULT_HISTORY_SIZE
 163  * network size estimates (matching bits, actually).
 164  */
 165 static unsigned int size_estimates[DEFAULT_HISTORY_SIZE];
 166
 167 /**
 168  * Array of size estimate messages.
 169  */
 170 static struct GNUNET_NSE_FloodMessage
 171     size_estimate_messages[DEFAULT_HISTORY_SIZE];
 172
 173 /**
 174  * Index of most recent estimate.
 175  */
 176 static unsigned int estimate_index;
 177
 178 /**
 179  * Task scheduled to send flood message.
 180  */
 181 static GNUNET_SCHEDULER_TaskIdentifier flood_task;
 182
 183 /**
 184  * Task to schedule flood message and update state.
 185  */
 186 static GNUNET_SCHEDULER_TaskIdentifier schedule_flood_task;
 187
 188 /**
 189  * Notification context, simplifies client broadcasts.
 190  */
 191 static struct GNUNET_SERVER_NotificationContext *nc;
 192
 193 /**
 194  * The previous major time.
 195  */
 196 static struct GNUNET_TIME_Absolute previous_timestamp;
 197
 198 /**
 199  * The next major time.
 200  */
 201 static struct GNUNET_TIME_Absolute next_timestamp;
 202
 203 /**
 204  * Base increment of time to add to send time.
 205  */
 206 static struct GNUNET_TIME_Relative increment;
 207
 208 /**
 209  * The current network size estimate message.
 210  */
 211 static struct GNUNET_NSE_ClientMessage current_estimate_message;
 212
 213 /**
 214  * The public key of this peer.
 215  */
 216 static struct GNUNET_CRYPTO_RsaPublicKeyBinaryEncoded my_public_key;
 217
 218 /**
 219  * The private key of this peer.
 220  */
 221 static struct GNUNET_CRYPTO_RsaPrivateKey *my_private_key;
 222
 223 /**
 224  * The peer identity of this peer.
 225  */
 226 static struct GNUNET_PeerIdentity my_identity;
 227
 228 /**
 229  * Our flood message, updated whenever a flood is sent.
 230  */
 231 static struct GNUNET_NSE_FloodMessage flood_message;
 232
 233 /**
 234  * Handler for START message from client, triggers an
 235  * immediate current network estimate notification.
 236  * Also, we remember the client for updates upon future
 237  * estimate measurements.
 238  *
 239  * @param cls unused
 240  * @param client who sent the message
 241  * @param message the message received
 242  */
 243 static void
 244 handle_start_message(void *cls, struct GNUNET_SERVER_Client *client,
 245                      const struct GNUNET_MessageHeader *message)
 246 {
 247   if ((ntohs (message->size) != sizeof(struct GNUNET_MessageHeader))
 248       || (ntohs (message->type) != GNUNET_MESSAGE_TYPE_NSE_START))
 249     return;
 250
 251 #if DEBUG_NSE
 252   GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG, "NSE",
 253       "Received START message from client\n");
 254 #endif
 255   GNUNET_SERVER_notification_context_add (nc, client);
 256   GNUNET_SERVER_receive_done (client, GNUNET_OK);
 257 }
 258
 259 /**
 260  * Called when core is ready to send a message we asked for
 261  * out to the destination.
 262  *
 263  * @param cls closure (NULL)
 264  * @param size number of bytes available in buf
 265  * @param buf where the callee should write the message
 266  * @return number of bytes written to buf
 267  */
 268 static size_t
 269 transmit_ready(void *cls, size_t size, void *buf)
 270 {
 271   struct NSEPeerEntry *peer_entry = cls;
 272   char *cbuf = buf;
 273
 274   size_t msize;
 275   peer_entry->th = NULL;
 276 #if DEBUG_NSE
 277   GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "%s: transmit_ready called\n",
 278       GNUNET_i2s (&my_identity));
 279 #endif
 280   if (buf == NULL) /* client disconnected */
 281     {
 282       GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
 283                   "%s: transmit_ready called (disconnect)\n",
 284                   GNUNET_i2s (&my_identity));
 285       return 0;
 286     }
 287
 288   if (peer_entry->pending_message == NULL)
 289     {
 290       GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
 291                   "%s: transmit_ready called (no message)\n",
 292                   GNUNET_i2s (&my_identity));
 293       return 0;
 294     }
 295
 296   msize = ntohs (peer_entry->pending_message->size);
 297   if (msize <= size)
 298     memcpy (cbuf, peer_entry->pending_message, msize);
 299 #if DEBUG_NSE
 300   GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
 301       "%s: transmit_ready called (transmit %d bytes)\n",
 302       GNUNET_i2s (&my_identity), msize);
 303 #endif
 304   return msize;
 305 }
 306
 307 /**
 308  * We sent on our flood message or one that we received
 309  * which was validated and closer than ours.  Update the
 310  * global list of recent messages and the average.  Also
 311  * re-broadcast the message to any clients.
 312  *
 313  * @param message the network flood message
 314  */
 315 static void
 316 update_network_size_estimate(struct GNUNET_NSE_FloodMessage *message)
 317 {
 318   unsigned int i;
 319   unsigned int count;
 320   double average;
 321   double std_dev;
 322   double diff;
 323
 324   size_estimates[estimate_index] = htonl (message->distance);
 325   memcpy (&size_estimate_messages[estimate_index], message,
 326           sizeof(struct GNUNET_NSE_FloodMessage));
 327
 328   count = 0;
 329   std_dev = 0.0;
 330   average = 0.0;
 331   for (i = 0; i < DEFAULT_HISTORY_SIZE; i++)
 332     {
 333       if (size_estimate_messages[i].distance != 0)
 334         {
 335 #if AVERAGE_SQUARE
 336           average += (1 << htonl (size_estimate_messages[i].distance));
 337 #else
 338           average += htonl (size_estimate_messages[i].distance);
 339 #endif
 340           count++;
 341         }
 342     }
 343
 344   if (count > 0)
 345     {
 346       average /= (double) count;
 347       for (i = 0; i < DEFAULT_HISTORY_SIZE; i++)
 348         {
 349           if (size_estimate_messages[i].distance != 0)
 350             {
 351 #if DEBUG_NSE
 352               GNUNET_log(GNUNET_ERROR_TYPE_WARNING, "%s: estimate %d %d\n", GNUNET_i2s(&my_identity), i, (1 << htonl(size_estimate_messages[i].distance)));
 353 #endif
 354 #if AVERAGE_SQUARE
 355               diff = average
 356                   - (1 << htonl (size_estimate_messages[i].distance));
 357 #else
 358               diff = average - htonl (size_estimate_messages[i].distance);
 359 #endif
 360               std_dev += diff * diff;
 361             }
 362         }
 363       std_dev /= count;
 364       std_dev = sqrt (std_dev);
 365       current_estimate_message.header.size
 366           = htons (sizeof(struct GNUNET_NSE_ClientMessage));
 367       current_estimate_message.header.type
 368           = htons (GNUNET_MESSAGE_TYPE_NSE_ESTIMATE);
 369 #if AVERAGE_SQUARE
 370       current_estimate_message.size_estimate = average;
 371       current_estimate_message.std_deviation = std_dev;
 372 #else
 373       current_estimate_message.size_estimate = pow(2, average);
 374       current_estimate_message.std_deviation = pow(2, std_dev);
 375 #endif
 376       /* Finally, broadcast the current estimate to all clients */
 377 #if DEBUG_NSE
 378       GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
 379           "%s: sending estimate %f -- %f to client\n",
 380           GNUNET_i2s (&my_identity),
 381           average,
 382           std_dev);
 383 #endif
 384       GNUNET_SERVER_notification_context_broadcast (
 385                                                     nc,
 386                                                     &current_estimate_message.header,
 387                                                     GNUNET_NO);
 388
 389       GNUNET_STATISTICS_set (stats, "Current network size estimate",
 390                              (uint64_t) average, GNUNET_NO);
 391     }
 392 }
 393
 394 static void
 395 send_flood_message(void *cls, const struct GNUNET_SCHEDULER_TaskContext * tc);
 396
 397 /**
 398  * Schedule a flood message to be sent.
 399  *
 400  * @param cls unused
 401  * @param tc context for this message
 402  *
 403  * This should be called on startup,
 404  * when a valid flood message is received (and
 405  * the next send flood message hasn't been
 406  * scheduled yet) and when this peer sends
 407  * a valid flood message.  As such, there should
 408  * always be a message scheduled to be sent.
 409  */
 410 static void
 411 schedule_flood_message(void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
 412 {
 413   GNUNET_HashCode timestamp_hash;
 414   struct GNUNET_TIME_Absolute curr_time;
 415   struct GNUNET_TIME_Relative offset;
 416   unsigned int matching_bits;
 417   double millisecond_offset;
 418
 419   schedule_flood_task = GNUNET_SCHEDULER_NO_TASK;
 420   if ((tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN) != 0)
 421     return;
 422
 423   GNUNET_assert(flood_task == GNUNET_SCHEDULER_NO_TASK);
 424
 425   if (0 != GNUNET_TIME_absolute_get_remaining (next_timestamp).rel_value)
 426     {
 427       GNUNET_break(0); /* Shouldn't ever happen! */
 428       schedule_flood_task
 429           = GNUNET_SCHEDULER_add_delayed (
 430                                           GNUNET_TIME_absolute_get_remaining (
 431                                                                               next_timestamp),
 432                                           &schedule_flood_message, NULL);
 433     }
 434
 435   /* Get the current UTC time */
 436   curr_time = GNUNET_TIME_absolute_get ();
 437   /* Find the previous interval start time */
 438   previous_timestamp.abs_value = (curr_time.abs_value / GNUNET_NSE_INTERVAL)
 439       * GNUNET_NSE_INTERVAL;
 440   /* Find the next interval start time */
 441   next_timestamp.abs_value = previous_timestamp.abs_value + GNUNET_NSE_INTERVAL;
 442 #if DEBUG_NSE
 443   GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
 444       "%s: curr_time %lu, prev timestamp %lu, next timestamp %lu\n",
 445       GNUNET_i2s (&my_identity), curr_time.abs_value,
 446       previous_timestamp.abs_value, next_timestamp.abs_value);
 447 #endif
 448   GNUNET_CRYPTO_hash (&next_timestamp.abs_value,
 449                       sizeof(next_timestamp.abs_value), &timestamp_hash);
 450   matching_bits = GNUNET_CRYPTO_hash_matching_bits (&timestamp_hash,
 451                                                     &my_identity.hashPubKey);
 452
 453   flood_message.header.size = htons (sizeof(struct GNUNET_NSE_FloodMessage));
 454   flood_message.header.type = htons (GNUNET_MESSAGE_TYPE_NSE_P2P_FLOOD);
 455   flood_message.purpose.purpose = htonl (GNUNET_SIGNATURE_PURPOSE_NSE_SEND);
 456   flood_message.purpose.size = htonl (sizeof(struct GNUNET_NSE_FloodMessage)
 457       - sizeof(struct GNUNET_MessageHeader) - sizeof(flood_message.signature));
 458   flood_message.distance = htonl (matching_bits);
 459   flood_message.timestamp = GNUNET_TIME_absolute_hton (next_timestamp);
 460   memcpy (&flood_message.pkey, &my_public_key,
 461           sizeof(struct GNUNET_CRYPTO_RsaPublicKeyBinaryEncoded));
 462   flood_message.proof_of_work = htonl (0);
 463   GNUNET_CRYPTO_rsa_sign (my_private_key, &flood_message.purpose,
 464                           &flood_message.signature);
 465
 466   /*S + f/2 - (f / pi) * (atan(x - p'))*/
 467
 468   // S is next_timestamp
 469   // f is frequency (GNUNET_NSE_INTERVAL)
 470   // x is matching_bits
 471   // p' is current_size_estimate
 472   millisecond_offset = ((double) GNUNET_NSE_INTERVAL / (double) 2)
 473       - ((GNUNET_NSE_INTERVAL / M_PI) * atan (matching_bits
 474           - current_size_estimate));
 475 #if DEBUG_NSE
 476   GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
 477       "%s: id matches %d bits, offset is %lu\n\n",
 478       GNUNET_i2s (&my_identity), matching_bits,
 479       (uint64_t) millisecond_offset);
 480 #endif
 481   /* Stop initial call from incrementing */
 482   if (size_estimate_messages[estimate_index].distance != 0)
 483     estimate_index += 1;
 484
 485   if (estimate_index >= DEFAULT_HISTORY_SIZE)
 486     estimate_index = 0;
 487
 488   if (millisecond_offset < curr_time.abs_value - previous_timestamp.abs_value)
 489     offset.rel_value = 0;
 490   else
 491     offset.rel_value = (uint64_t) millisecond_offset + curr_time.abs_value
 492         - previous_timestamp.abs_value;
 493 #if DEBUG_NSE
 494   GNUNET_log (
 495       GNUNET_ERROR_TYPE_WARNING,
 496       "%s: milliseconds until next timestamp %lu, sending flood in %lu\n",
 497       GNUNET_i2s (&my_identity),
 498       GNUNET_TIME_absolute_get_remaining (next_timestamp).rel_value,
 499       offset.rel_value);
 500 #endif
 501   flood_task = GNUNET_SCHEDULER_add_delayed (offset, &send_flood_message, NULL);
 502
 503 }
 504
 505 #if VERIFY_CRYPTO
 506 /**
 507  * Check whether the given public key
 508  * and integer are a valid proof of work.
 509  *
 510  * @param pkey the public key
 511  * @param val the integer
 512  * @param want the number of trailing zeroes
 513  *
 514  * @return GNUNET_YES if valid, GNUNET_NO if not
 515  */
 516 static int check_proof_of_work(struct GNUNET_CRYPTO_RsaPublicKeyBinaryEncoded pkey, uint64_t val, unsigned int want)
 517   {
 518
 519     return GNUNET_YES;
 520   }
 521
 522 /**
 523  * Count the trailing zeroes in hash.
 524  *
 525  * @param hash
 526  *
 527  * @return the number of trailing zero bits.
 528  */
 529 static unsigned int count_trailing_zeroes(GNUNET_HashCode *hash)
 530   {
 531     unsigned int hash_count;
 532
 533     hash_count = sizeof(GNUNET_HashCode) * 8;
 534     while ((0 == GNUNET_CRYPTO_hash_get_bit(hash, hash_count)))
 535     hash_count--;
 536     return (sizeof(GNUNET_HashCode) * 8) - hash_count;
 537   }
 538
 539 /**
 540  * Given a public key, find an integer such that
 541  * the hash of the key concatenated with the integer
 542  * has <param>want</param> trailing 0 bits.
 543  *
 544  * @param pkey the public key
 545  * @param want the number of trailing 0 bits
 546  *
 547  * @return 64 bit number that satisfies the
 548  *         requirements
 549  *
 550  * FIXME: use pointer and return GNUNET_YES or
 551  *        GNUNET_NO in case no such number works?
 552  */
 553 static uint64_t find_proof_of_work(struct GNUNET_CRYPTO_RsaPublicKeyBinaryEncoded *pkey, unsigned int want)
 554   {
 555     uint64_t counter;
 556     static char buf[sizeof(struct GNUNET_CRYPTO_RsaPublicKeyBinaryEncoded) + sizeof(uint64_t)];
 557     unsigned int data_size;
 558     static GNUNET_HashCode result;
 559
 560     data_size = sizeof(struct GNUNET_CRYPTO_RsaPublicKeyBinaryEncoded) + sizeof(uint64_t);
 561     memcpy(buf, pkey, sizeof(struct GNUNET_CRYPTO_RsaPublicKeyBinaryEncoded));
 562     counter = 0;
 563     while (counter != (uint64_t)-1)
 564       {
 565         memcpy(&buf[sizeof(struct GNUNET_CRYPTO_RsaPublicKeyBinaryEncoded)], &counter, sizeof(uint64_t));
 566         GNUNET_CRYPTO_hash(buf, data_size, &result);
 567         if (want == count_trailing_zeroes(&result)) /* Found good proof of work! */
 568         break;
 569         counter++;
 570       }
 571     if (counter < (uint64_t)-1)
 572     return counter; /* Found valid proof of work */
 573     else
 574     return 0; /* Did not find valid proof of work */
 575   }
 576
 577 /**
 578  * An incoming flood message has been received which claims
 579  * to have more bits matching than any we know in this time
 580  * period.  Verify the signature and/or proof of work.
 581  *
 582  * @param incoming_flood the message to verify
 583  *
 584  * @return GNUNET_YES if the message is verified
 585  *         GNUNET_NO if the key/signature don't verify
 586  */
 587 static int verify_message_crypto(struct GNUNET_NSE_FloodMessage *incoming_flood)
 588   {
 589     int ret;
 590     if (GNUNET_OK == (ret
 591             = GNUNET_CRYPTO_rsa_verify (GNUNET_SIGNATURE_PURPOSE_NSE_SEND,
 592                 &incoming_flood->purpose,
 593                 &incoming_flood->signature,
 594                 &incoming_flood->pkey)))
 595     return GNUNET_YES;
 596
 597     return GNUNET_NO;
 598   }
 599 #endif
 600
 601 /**
 602  * Core handler for size estimate flooding messages.
 603  *
 604  * @param cls closure unused
 605  * @param message message
 606  * @param peer peer identity this message is from (ignored)
 607  * @param atsi performance data (ignored)
 608  *
 609  */
 610 static int
 611 handle_p2p_size_estimate(void *cls, const struct GNUNET_PeerIdentity *peer,
 612                          const struct GNUNET_MessageHeader *message,
 613                          const struct GNUNET_TRANSPORT_ATS_Information *atsi)
 614 {
 615   struct GNUNET_NSE_FloodMessage *incoming_flood;
 616   struct GNUNET_TIME_Absolute curr_time;
 617   uint64_t drift;
 618
 619 #if DEBUG_NSE
 620   GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "%s: received flood message!\n",
 621       GNUNET_i2s (&my_identity));
 622 #endif
 623   if (ntohs (message->size) != sizeof(struct GNUNET_NSE_FloodMessage))
 624     {
 625       GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "%s: bad message size!\n",
 626                   GNUNET_i2s (&my_identity));
 627       return GNUNET_NO;
 628     }
 629
 630   GNUNET_STATISTICS_update (stats, "# flood messages received", 1, GNUNET_NO);
 631   incoming_flood = (struct GNUNET_NSE_FloodMessage *) message;
 632   if (ntohl (incoming_flood->distance)
 633       <= ntohl (size_estimate_messages[estimate_index].distance)) /* Not closer than our most recent message */
 634     {
 635 #if DEBUG_NSE
 636       GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
 637           "%s: distance %d not greater than %d, discarding\n",
 638           GNUNET_i2s (&my_identity), ntohl (incoming_flood->distance),
 639           ntohl (size_estimate_messages[estimate_index].distance));
 640 #endif
 641       GNUNET_STATISTICS_update (stats,
 642                                 "# flood messages discarded (had closer)", 1,
 643                                 GNUNET_NO);
 644       return GNUNET_OK;
 645     }
 646
 647   curr_time = GNUNET_TIME_absolute_get ();
 648   if (curr_time.abs_value
 649       > GNUNET_TIME_absolute_ntoh (incoming_flood->timestamp).abs_value)
 650     drift = curr_time.abs_value
 651         - GNUNET_TIME_absolute_ntoh (incoming_flood->timestamp).abs_value;
 652   else
 653     drift = GNUNET_TIME_absolute_ntoh (incoming_flood->timestamp).abs_value
 654         - curr_time.abs_value;
 655
 656   if (drift > GNUNET_NSE_DRIFT_TOLERANCE)
 657     {
 658       GNUNET_STATISTICS_update (
 659                                 stats,
 660                                 "# flood messages discarded (clock skew too high)",
 661                                 1, GNUNET_NO);
 662       return GNUNET_OK;
 663     }
 664
 665 #if VERIFY_CRYPTO
 666   if (GNUNET_YES != verify_message_crypto(incoming_flood))
 667     {
 668       GNUNET_STATISTICS_update (stats,
 669           "# flood messages discarded (bad crypto)",
 670           1, GNUNET_NO);
 671       return GNUNET_OK;
 672     }
 673 #endif
 674
 675   /* Have a new, better size estimate! */
 676   update_network_size_estimate (incoming_flood);
 677
 678   if (flood_task != GNUNET_SCHEDULER_NO_TASK)
 679     {
 680 #if DEBUG_NSE
 681       GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "%s: received closer message, canceling my flood task!\n", GNUNET_i2s(&my_identity));
 682 #endif
 683       GNUNET_SCHEDULER_cancel (flood_task);
 684       flood_task = GNUNET_SCHEDULER_NO_TASK;
 685     }
 686
 687   /** Commenting out prevents forwarding of messages */
 688 #if DO_FORWARD
 689   GNUNET_SCHEDULER_add_now(&send_flood_message, &size_estimate_messages[estimate_index]);
 690 #endif
 691   if (schedule_flood_task != GNUNET_SCHEDULER_NO_TASK)
 692     GNUNET_SCHEDULER_cancel (schedule_flood_task);
 693
 694   schedule_flood_task
 695       = GNUNET_SCHEDULER_add_delayed (
 696                                       GNUNET_TIME_absolute_get_remaining (
 697                                                                           next_timestamp),
 698                                       &schedule_flood_message, NULL);
 699
 700   return GNUNET_OK;
 701 }
 702
 703 /**
 704  * Send a flood message.
 705  *
 706  * If we've gotten here, it means either we haven't received
 707  * a network size estimate message closer than ours, or
 708  * we need to forward a message we received which was closer
 709  * than ours.
 710  */
 711 static void
 712 send_flood_message(void *cls, const struct GNUNET_SCHEDULER_TaskContext * tc)
 713 {
 714   struct NSEPeerEntry *peer_entry;
 715   struct GNUNET_NSE_FloodMessage *to_send;
 716
 717   if (cls == NULL) /* Means we are sending our OWN flood message */
 718     to_send = &flood_message;
 719   else
 720     /* Received a message from another peer that should be forwarded */
 721     to_send = (struct GNUNET_NSE_FloodMessage *) cls;
 722
 723   flood_task = GNUNET_SCHEDULER_NO_TASK;
 724   if ((tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN) != 0)
 725     return;
 726 #if DEBUG_NSE
 727   GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
 728       "%s: my time has come, sending flood message of size %d!\n",
 729       GNUNET_i2s (&my_identity), ntohs (to_send->header.size));
 730 #endif
 731   peer_entry = peers_head;
 732
 733   while (peer_entry != NULL)
 734     {
 735       peer_entry->pending_message = &to_send->header;
 736       peer_entry->th
 737           = GNUNET_CORE_notify_transmit_ready (
 738                                                coreAPI,
 739                                                GNUNET_NO,
 740                                                DEFAULT_NSE_PRIORITY,
 741                                                GNUNET_TIME_absolute_get_remaining (
 742                                                                                    next_timestamp),
 743                                                &peer_entry->id,
 744                                                ntohs (to_send->header.size),
 745                                                &transmit_ready, peer_entry);
 746       if (peer_entry->th == NULL)
 747         GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
 748                     "%s: transmit handle is null!\n", GNUNET_i2s (&my_identity));
 749 #if DEBUG_NSE
 750       GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
 751           "%s: Sending flood message (distance %d) to %s!\n",
 752           GNUNET_i2s (&my_identity), ntohl (to_send->distance),
 753           GNUNET_h2s (&peer_entry->id.hashPubKey));
 754 #endif
 755       peer_entry = peer_entry->next;
 756     }
 757
 758   if (cls == NULL) /* Need to update our size estimate */
 759     {
 760       update_network_size_estimate (to_send);
 761       GNUNET_STATISTICS_update (stats, "# flood messages sent", 1, GNUNET_NO);
 762     }
 763   else
 764     GNUNET_STATISTICS_update (stats, "# flood messages forwarded", 1, GNUNET_NO);
 765
 766 #if DEBUG_NSE
 767   GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
 768       "%s: scheduling schedule_flood_message in %lu\n",
 769       GNUNET_i2s (&my_identity),
 770       GNUNET_TIME_absolute_get_remaining (next_timestamp).rel_value);
 771 #endif
 772   if (schedule_flood_task != GNUNET_SCHEDULER_NO_TASK)
 773     GNUNET_SCHEDULER_cancel (schedule_flood_task);
 774
 775   schedule_flood_task
 776       = GNUNET_SCHEDULER_add_delayed (
 777                                       GNUNET_TIME_absolute_get_remaining (
 778                                                                           next_timestamp),
 779                                       &schedule_flood_message, NULL);
 780 }
 781
 782 /**
 783  * Method called whenever a peer connects.
 784  *
 785  * @param cls closure
 786  * @param peer peer identity this notification is about
 787  * @param atsi performance data
 788  */
 789 static void
 790 handle_core_connect(void *cls, const struct GNUNET_PeerIdentity *peer,
 791                     const struct GNUNET_TRANSPORT_ATS_Information *atsi)
 792 {
 793   struct NSEPeerEntry *peer_entry;
 794
 795   if (0 == (memcmp (peer, &my_identity, sizeof(struct GNUNET_PeerIdentity))))
 796     return; /* Do not connect to self... */
 797
 798   peer_entry = GNUNET_malloc(sizeof(struct NSEPeerEntry));
 799   memcpy (&peer_entry->id, peer, sizeof(struct GNUNET_PeerIdentity));
 800   GNUNET_CONTAINER_DLL_insert(peers_head, peers_tail, peer_entry);
 801 }
 802
 803 /**
 804  * Method called whenever a peer disconnects.
 805  *
 806  * @param cls closure
 807  * @param peer peer identity this notification is about
 808  */
 809 static void
 810 handle_core_disconnect(void *cls, const struct GNUNET_PeerIdentity *peer)
 811 {
 812   struct NSEPeerEntry *pos;
 813
 814   if (0 == (memcmp (peer, &my_identity, sizeof(struct GNUNET_PeerIdentity))))
 815     return; /* Ignore disconnect from self... */
 816
 817   pos = peers_head;
 818   while ((NULL != pos) && (0 != memcmp (&pos->id, peer,
 819                                         sizeof(struct GNUNET_PeerIdentity))))
 820     pos = pos->next;
 821   if (pos == NULL)
 822     {
 823       GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
 824                   "Received disconnect before connect!\n");
 825       GNUNET_break(0); /* Should never receive a disconnect message for a peer we don't know about... */
 826       return;
 827     }
 828
 829   /* TODO: decide whether to copy the message, or always use the static pointer */
 830 #if TODO
 831   if (pos->pending_message != NULL)
 832   GNUNET_free(pos->pending_message);
 833 #endif
 834
 835   if (pos->th != NULL)
 836     GNUNET_CORE_notify_transmit_ready_cancel (pos->th);
 837   GNUNET_CONTAINER_DLL_remove(peers_head, peers_tail, pos);
 838   GNUNET_free(pos);
 839 }
 840
 841 /**
 842  * A client disconnected. Remove it from the
 843  * global DLL of clients.
 844  *
 845  * @param cls closure, NULL
 846  * @param client identification of the client
 847  */
 848 static void
 849 handle_client_disconnect(void *cls, struct GNUNET_SERVER_Client* client)
 850 {
 851   struct ClientListEntry *cle;
 852
 853   while (NULL != (cle = cle_head))
 854     cle = cle->next;
 855
 856   if (cle != NULL)
 857     {
 858       GNUNET_SERVER_client_drop (cle->client);
 859       GNUNET_CONTAINER_DLL_remove(cle_head,
 860           cle_tail,
 861           cle);
 862       GNUNET_free(cle);
 863     }
 864   if (coreAPI != NULL)
 865     {
 866       GNUNET_CORE_disconnect (coreAPI);
 867       coreAPI = NULL;
 868     }
 869 }
 870
 871 /**
 872  * Task run during shutdown.
 873  *
 874  * @param cls unused
 875  * @param tc unused
 876  */
 877 static void
 878 shutdown_task(void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
 879 {
 880   struct ClientListEntry *cle;
 881
 882   if (flood_task != GNUNET_SCHEDULER_NO_TASK)
 883     GNUNET_SCHEDULER_cancel (flood_task);
 884   GNUNET_SERVER_notification_context_destroy (nc);
 885   nc = NULL;
 886   while (NULL != (cle = cle_head))
 887     {
 888       GNUNET_SERVER_client_drop (cle->client);
 889       GNUNET_CONTAINER_DLL_remove (cle_head,
 890           cle_tail,
 891           cle);
 892       GNUNET_free (cle);
 893     }
 894
 895   if (coreAPI != NULL)
 896     {
 897       GNUNET_CORE_disconnect (coreAPI);
 898       coreAPI = NULL;
 899     }
 900
 901   if (stats != NULL)
 902     GNUNET_STATISTICS_destroy (stats, GNUNET_NO);
 903
 904 }
 905
 906 /**
 907  * Called on core init/fail.
 908  *
 909  * @param cls service closure
 910  * @param server handle to the server for this service
 911  * @param identity the public identity of this peer
 912  * @param publicKey the public key of this peer
 913  */
 914 void
 915 core_init(void *cls, struct GNUNET_CORE_Handle *server,
 916           const struct GNUNET_PeerIdentity *identity,
 917           const struct GNUNET_CRYPTO_RsaPublicKeyBinaryEncoded *publicKey)
 918 {
 919   struct GNUNET_TIME_Absolute curr_time;
 920   if (server == NULL)
 921     {
 922 #if DEBUG_NSE
 923       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "%s: Connection to core FAILED!\n",
 924           "nse", GNUNET_i2s (identity));
 925 #endif
 926       GNUNET_SCHEDULER_add_now (&shutdown_task, NULL);
 927       return;
 928     }
 929
 930   /* Copy our identity so we can use it */
 931   memcpy (&my_identity, identity, sizeof(struct GNUNET_PeerIdentity));
 932   /* Copy our public key for inclusion in flood messages */
 933   memcpy (&my_public_key, publicKey,
 934           sizeof(struct GNUNET_CRYPTO_RsaPublicKeyBinaryEncoded));
 935
 936   if (flood_task != GNUNET_SCHEDULER_NO_TASK)
 937     GNUNET_SCHEDULER_cancel (flood_task);
 938
 939   /* Get the current UTC time */
 940   curr_time = GNUNET_TIME_absolute_get ();
 941   /* Find the previous interval start time */
 942   previous_timestamp.abs_value = (curr_time.abs_value / GNUNET_NSE_INTERVAL)
 943       * GNUNET_NSE_INTERVAL;
 944   /* Find the next interval start time */
 945   next_timestamp.abs_value = previous_timestamp.abs_value + GNUNET_NSE_INTERVAL;
 946
 947 #if DEBUG_NSE
 948   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 949       "%s: Core connection initialized, I am peer: %s, scheduling flood task in %lu\n", "nse",
 950       GNUNET_i2s (identity), GNUNET_TIME_absolute_get_remaining(next_timestamp));
 951 #endif
 952   /* FIXME: In production, we'd likely want to do this immediately, but in test-beds it causes stupid behavior */
 953   if (schedule_flood_task != GNUNET_SCHEDULER_NO_TASK)
 954     GNUNET_SCHEDULER_cancel (schedule_flood_task);
 955   schedule_flood_task
 956       = GNUNET_SCHEDULER_add_delayed (
 957                                       GNUNET_TIME_absolute_get_remaining (
 958                                                                           next_timestamp),
 959                                       &schedule_flood_message, NULL);
 960
 961   GNUNET_SERVER_notification_context_broadcast (
 962                                                 nc,
 963                                                 &current_estimate_message.header,
 964                                                 GNUNET_NO);
 965 }
 966
 967 /**
 968  * Handle network size estimate clients.
 969  *
 970  * @param cls closure
 971  * @param server the initialized server
 972  * @param c configuration to use
 973  */
 974 static void
 975 run(void *cls, struct GNUNET_SERVER_Handle *server,
 976     const struct GNUNET_CONFIGURATION_Handle *c)
 977 {
 978   char *keyfile;
 979   static const struct GNUNET_SERVER_MessageHandler handlers[] =
 980     {
 981       { &handle_start_message, NULL, GNUNET_MESSAGE_TYPE_NSE_START, 0 },
 982       { NULL, NULL, 0, 0 } };
 983
 984   static const struct GNUNET_CORE_MessageHandler core_handlers[] =
 985     {
 986       { &handle_p2p_size_estimate, GNUNET_MESSAGE_TYPE_NSE_P2P_FLOOD, 0 },
 987       { NULL, 0, 0 } };
 988
 989   cfg = c;
 990
 991   if (GNUNET_OK
 992       != GNUNET_CONFIGURATION_get_value_filename (c, "GNUNETD", "HOSTKEY",
 993                                                   &keyfile))
 994     {
 995       GNUNET_log (GNUNET_ERROR_TYPE_ERROR, _
 996       ("NSE service is lacking key configuration settings.  Exiting.\n"));
 997       GNUNET_SCHEDULER_shutdown ();
 998       return;
 999     }
1000
1001   my_private_key = GNUNET_CRYPTO_rsa_key_create_from_file (keyfile);
1002   GNUNET_free (keyfile);
1003   if (my_private_key == NULL)
1004     {
1005       GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
1006                   _("NSE Service could not access hostkey.  Exiting.\n"));
1007       GNUNET_SCHEDULER_shutdown ();
1008       return;
1009     }
1010
1011   GNUNET_SERVER_add_handlers (server, handlers);
1012   nc = GNUNET_SERVER_notification_context_create (server, 16);
1013   GNUNET_SERVER_disconnect_notify (server, &handle_client_disconnect, NULL);
1014
1015   flood_task = GNUNET_SCHEDULER_NO_TASK;
1016   /** Connect to core service and register core handlers */
1017   coreAPI = GNUNET_CORE_connect (cfg, /* Main configuration */
1018   DEFAULT_CORE_QUEUE_SIZE, /* queue size */
1019   NULL, /* Closure passed to functions */
1020   &core_init, /* Call core_init once connected */
1021   &handle_core_connect, /* Handle connects */
1022   &handle_core_disconnect, /* Handle disconnects */
1023   NULL, /* Do we care about "status" updates? */
1024   NULL, /* Don't want notified about all incoming messages */
1025   GNUNET_NO, /* For header only inbound notification */
1026   NULL, /* Don't want notified about all outbound messages */
1027   GNUNET_NO, /* For header only outbound notification */
1028   core_handlers); /* Register these handlers */
1029
1030   if (coreAPI == NULL)
1031     {
1032       GNUNET_SCHEDULER_add_now (&shutdown_task, NULL);
1033       return;
1034     }
1035
1036   stats = GNUNET_STATISTICS_create ("NSE", cfg);
1037
1038   increment
1039       = GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MILLISECONDS,
1040                                        GNUNET_NSE_INTERVAL
1041                                            / (sizeof(GNUNET_HashCode) * 8));
1042   /* Set we have no idea defaults for network size estimate */
1043   current_size_estimate = 0.0;
1044   current_std_dev = NAN;
1045   size_estimates[estimate_index] = 0;
1046   current_estimate_message.header.size
1047       = htons (sizeof(struct GNUNET_NSE_ClientMessage));
1048   current_estimate_message.header.type
1049       = htons (GNUNET_MESSAGE_TYPE_NSE_ESTIMATE);
1050   current_estimate_message.size_estimate = current_size_estimate;
1051   current_estimate_message.std_deviation = current_std_dev;
1052   GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_FOREVER_REL, &shutdown_task,
1053                                 NULL);
1054 }
1055
1056 /**
1057  * The main function for the statistics service.
1058  *
1059  * @param argc number of arguments from the command line
1060  * @param argv command line arguments
1061  * @return 0 ok, 1 on error
1062  */
1063 int
1064 main(int argc, char * const *argv)
1065 {
1066   return (GNUNET_OK == GNUNET_SERVICE_run (argc, argv, "nse",
1067                                            GNUNET_SERVICE_OPTION_NONE, &run,
1068                                            NULL)) ? 0 : 1;
1069 }
1070
1071 /* End of gnunet-service-nse.c */
1072