- Added logging of peers to NSE service
[oweals/gnunet.git] / src / nse / gnunet-service-nse.c
index 543e4794aed97b8807bb19cc999bd0f70cc4f29e..d72670cb55cd5e15a381534d6c54214114779535 100644 (file)
 /**
  * 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.
@@ -303,43 +298,72 @@ 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);
@@ -412,15 +436,14 @@ 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 - 1);
+  i = (uint32_t) (d / (double) (hop_count_max + 1));
+  ret.rel_value = GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK, i + 1);
   return ret;
 #else
   return GNUNET_TIME_UNIT_ZERO;
@@ -550,6 +573,13 @@ transmit_ready (void *cls, size_t size, void *buf)
                               1, GNUNET_NO);
     return 0;
   }
+  if (ntohs (size_estimate_messages[idx].header.size) == 0)
+  {
+    GNUNET_STATISTICS_update (stats,
+                              "# flood messages not generated (lack of history)",
+                              1, GNUNET_NO);
+    return 0;
+  }
 #if DEBUG_NSE
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
               "In round %llu, sending to `%s' estimate with %u bits\n",
@@ -636,8 +666,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));
 }
 
 
@@ -687,6 +721,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)
   {
@@ -821,16 +857,8 @@ find_proof (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
       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++;
@@ -878,11 +906,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;
@@ -938,12 +967,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;
@@ -1094,10 +1124,12 @@ handle_p2p_size_estimate (void *cls, const struct GNUNET_PeerIdentity *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;
 
@@ -1107,11 +1139,14 @@ handle_core_connect (void *cls, const struct GNUNET_PeerIdentity *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,
                                     peer_entry);
+  GNUNET_STATISTICS_update (stats, "# peers", 1, GNUNET_NO);
 }
 
 
@@ -1147,6 +1182,7 @@ handle_core_disconnect (void *cls, const struct GNUNET_PeerIdentity *peer)
     pos->th = NULL;
   }
   GNUNET_free (pos);
+  GNUNET_STATISTICS_update (stats, "# peers", -1, GNUNET_NO);
 }
 
 
@@ -1211,16 +1247,13 @@ 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)
   {
@@ -1239,16 +1272,16 @@ core_init (void *cls, struct GNUNET_CORE_Handle *server,
       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++)
+  estimate_index = HISTORY_SIZE - 1;
+  estimate_count = 0;
+  if (GNUNET_YES == check_proof_of_work (&my_public_key, my_proof))
   {
     prev_time.abs_value =
-        current_timestamp.abs_value - (HISTORY_SIZE - i -
+        current_timestamp.abs_value - (estimate_index -
                                        1) * gnunet_nse_interval.rel_value;
-    setup_flood_message (i, prev_time);
+    setup_flood_message (estimate_index, prev_time);
+    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,
@@ -1357,12 +1390,10 @@ run (void *cls, struct GNUNET_SERVER_Handle *server,
   nc = GNUNET_SERVER_notification_context_create (server, 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 */
+                                 1, 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 */