optimize mqm_head scans by avoiding constantly scanning over definitively non-ready...
[oweals/gnunet.git] / src / cadet / gnunet-service-cadet-new_core.c
index 6b5a02dc15da46ea157bf22667888c40beeb9b17..25ffcb3cee88ef92b726098e65fb6e40d9a160d6 100644 (file)
@@ -27,6 +27,7 @@
  * All functions in this file should use the prefix GCO (Gnunet Cadet cOre (bottom))
  *
  * TODO:
+ * - properly implement GLOBAL message buffer, instead of per-route buffers
  * - Optimization: given BROKEN messages, destroy paths (?)
  */
 #include "platform.h"
@@ -36,6 +37,7 @@
 #include "gnunet-service-cadet-new_connection.h"
 #include "gnunet-service-cadet-new_tunnels.h"
 #include "gnunet_core_service.h"
+#include "gnunet_statistics_service.h"
 #include "cadet_protocol.h"
 
 
@@ -122,6 +124,10 @@ struct CadetRoute
    */
   struct GNUNET_TIME_Absolute last_use;
 
+  /**
+   * Position of this route in the #route_heap.
+   */
+  struct GNUNET_CONTAINER_HeapNode *hn;
 };
 
 
@@ -135,6 +141,21 @@ static struct GNUNET_CORE_Handle *core;
  */
 static struct GNUNET_CONTAINER_MultiShortmap *routes;
 
+/**
+ * Heap of routes, MIN-sorted by last activity.
+ */
+static struct GNUNET_CONTAINER_Heap *route_heap;
+
+/**
+ * Maximum number of concurrent routes this peer will support.
+ */
+static unsigned long long max_routes;
+
+/**
+ * Task to timeout routes.
+ */
+static struct GNUNET_SCHEDULER_Task *timeout_task;
+
 
 /**
  * Get the route corresponding to a hash.
@@ -186,6 +207,9 @@ route_message (struct CadetPeer *prev,
                   env);
     return;
   }
+  route->last_use = GNUNET_TIME_absolute_get ();
+  GNUNET_CONTAINER_heap_update_cost (route->hn,
+                                     route->last_use.abs_value_us);
   dir = (prev == route->prev.hop) ? &route->next : &route->prev;
   if (GNUNET_YES == dir->is_ready)
   {
@@ -210,7 +234,11 @@ route_message (struct CadetPeer *prev,
          GCP_2s (prev),
          GNUNET_i2s (GCP_get_id (dir->hop)),
          GNUNET_sh2s (&cid->connection_of_tunnel));
-    GNUNET_assert (dir->out_rpos == dir->out_wpos);
+    GNUNET_STATISTICS_update (stats,
+                              "# messages dropped due to full buffer",
+                              1,
+                              GNUNET_NO);
+  GNUNET_assert (dir->out_rpos == dir->out_wpos);
     GNUNET_MQ_discard (env);
     dir->out_rpos++;
     if (ROUTE_BUFFER_SIZE == dir->out_rpos)
@@ -289,6 +317,8 @@ destroy_route (struct CadetRoute *route)
        GNUNET_i2s  (GCP_get_id (route->prev.hop)),
        GNUNET_i2s2 (GCP_get_id (route->next.hop)),
        GNUNET_sh2s (&route->cid.connection_of_tunnel));
+  GNUNET_assert (route ==
+                 GNUNET_CONTAINER_heap_remove_node (route->hn));
   destroy_direction (&route->prev);
   destroy_direction (&route->next);
   GNUNET_free (route);
@@ -312,6 +342,8 @@ send_broken (struct RouteDirection *target,
   struct GNUNET_MQ_Envelope *env;
   struct GNUNET_CADET_ConnectionBrokenMessage *bm;
 
+  if (NULL == target->mqm)
+    return; /* Can't send notification, connection is down! */
   LOG (GNUNET_ERROR_TYPE_DEBUG,
        "Notifying %s about BROKEN route at %s-%s of connection %s\n",
        GCP_2s (target->hop),
@@ -326,12 +358,56 @@ send_broken (struct RouteDirection *target,
     bm->peer1 = *peer1;
   if (NULL != peer2)
     bm->peer2 = *peer2;
+
   GCP_request_mq_cancel (target->mqm,
                          env);
   target->mqm = NULL;
 }
 
 
+/**
+ * Function called to check if any routes have timed out, and if
+ * so, to clean them up.  Finally, schedules itself again at the
+ * earliest time where there might be more work.
+ *
+ * @param cls NULL
+ */
+static void
+timeout_cb (void *cls)
+{
+  struct CadetRoute *r;
+  struct GNUNET_TIME_Relative linger;
+  struct GNUNET_TIME_Absolute exp;
+
+  timeout_task = NULL;
+  linger = GNUNET_TIME_relative_multiply (keepalive_period,
+                                          3);
+  while (NULL != (r = GNUNET_CONTAINER_heap_peek (route_heap)))
+  {
+    exp = GNUNET_TIME_absolute_add (r->last_use,
+                                    linger);
+    if (0 != GNUNET_TIME_absolute_get_duration (exp).rel_value_us)
+    {
+      /* Route not yet timed out, wait until it does. */
+      timeout_task = GNUNET_SCHEDULER_add_at (exp,
+                                              &timeout_cb,
+                                              NULL);
+      return;
+    }
+    send_broken (&r->prev,
+                 &r->cid,
+                 NULL,
+                 NULL);
+    send_broken (&r->next,
+                 &r->cid,
+                 NULL,
+                 NULL);
+    destroy_route (r);
+  }
+  /* No more routes left, so no need for a #timeout_task */
+}
+
+
 /**
  * Function called when the message queue to the previous hop
  * becomes available/unavailable.  We expect this function to
@@ -399,6 +475,35 @@ dir_init (struct RouteDirection *dir,
 }
 
 
+/**
+ * We could not create the desired route.  Send a
+ * #GNUNET_MESSAGE_TYPE_CADET_CONNECTION_BROKEN
+ * message to @a target.
+ *
+ * @param target who should receive the message
+ * @param cid identifier of the connection/route that failed
+ * @param failure_at neighbour with which we failed to route,
+ *        or NULL.
+ */
+static void
+send_broken_without_mqm (struct CadetPeer *target,
+                         const struct GNUNET_CADET_ConnectionTunnelIdentifier *cid,
+                         const struct GNUNET_PeerIdentity *failure_at)
+{
+  struct GNUNET_MQ_Envelope *env;
+  struct GNUNET_CADET_ConnectionBrokenMessage *bm;
+
+  env = GNUNET_MQ_msg (bm,
+                       GNUNET_MESSAGE_TYPE_CADET_CONNECTION_BROKEN);
+  bm->cid = *cid;
+  bm->peer1 = my_full_id;
+  if (NULL != failure_at)
+    bm->peer2 = *failure_at;
+  GCP_send_ooo (target,
+                env);
+}
+
+
 /**
  * Handle for #GNUNET_MESSAGE_TYPE_CADET_CONNECTION_CREATE
  *
@@ -443,7 +548,7 @@ handle_connection_create (void *cls,
   {
     /* Duplicate CREATE, pass it on, previous one might have been lost! */
     LOG (GNUNET_ERROR_TYPE_DEBUG,
-         "Passing on duplicate CREATE message on connection %s\n",
+         "Passing on duplicate CADET_CONNECTION_CREATE message on connection %s\n",
          GNUNET_sh2s (&msg->cid.connection_of_tunnel));
     route_message (sender,
                    &msg->cid,
@@ -462,24 +567,38 @@ handle_connection_create (void *cls,
     if (NULL != cc)
     {
       LOG (GNUNET_ERROR_TYPE_DEBUG,
-           "Received duplicate CREATE message on connection %s\n",
+           "Received duplicate CADET_CONNECTION_CREATE message on connection %s\n",
            GNUNET_sh2s (&msg->cid.connection_of_tunnel));
       GCC_handle_duplicate_create (cc);
       return;
     }
 
-    path = GCPP_get_path_from_route (path_length - 1,
-                                     pids);
     origin = GCP_get (&pids[0],
                       GNUNET_YES);
     LOG (GNUNET_ERROR_TYPE_DEBUG,
-         "Received CREATE message from %s via path %s for connection %s\n",
+         "Received CADET_CONNECTION_CREATE message from %s for connection %s, building inverse path\n",
          GCP_2s (origin),
-         GCPP_2s (path),
          GNUNET_sh2s (&msg->cid.connection_of_tunnel));
-    GCT_add_inbound_connection (GCT_create_tunnel (origin),
-                                &msg->cid,
-                                path);
+    path = GCPP_get_path_from_route (path_length - 1,
+                                     pids);
+    if (GNUNET_OK !=
+        GCT_add_inbound_connection (GCP_get_tunnel (origin,
+                                                    GNUNET_YES),
+                                    &msg->cid,
+                                    path))
+    {
+      /* Send back BROKEN: duplicate connection on the same path,
+         we will use the other one. */
+      LOG (GNUNET_ERROR_TYPE_DEBUG,
+           "Received CADET_CONNECTION_CREATE from %s for %s, but %s already has a connection. Sending BROKEN\n",
+           GCP_2s (sender),
+           GNUNET_sh2s (&msg->cid.connection_of_tunnel),
+           GCPP_2s (path));
+      send_broken_without_mqm (sender,
+                               &msg->cid,
+                               NULL);
+      return;
+    }
     return;
   }
   /* We are merely a hop on the way, check if we can support the route */
@@ -489,34 +608,39 @@ handle_connection_create (void *cls,
        (GNUNET_NO == GCP_has_core_connection (next)) )
   {
     /* unworkable, send back BROKEN notification */
-    struct GNUNET_MQ_Envelope *env;
-    struct GNUNET_CADET_ConnectionBrokenMessage *bm;
-
     LOG (GNUNET_ERROR_TYPE_DEBUG,
-         "Received CONNECTION_CREATE from %s for %s. Next hop %s:%u is down. Sending BROKEN\n",
+         "Received CADET_CONNECTION_CREATE from %s for %s. Next hop %s:%u is down. Sending BROKEN\n",
          GCP_2s (sender),
          GNUNET_sh2s (&msg->cid.connection_of_tunnel),
          GNUNET_i2s (&pids[off + 1]),
          off + 1);
-    env = GNUNET_MQ_msg (bm,
-                         GNUNET_MESSAGE_TYPE_CADET_CONNECTION_BROKEN);
-    bm->cid = msg->cid;
-    bm->peer1 = pids[off + 1];
-    bm->peer2 = my_full_id;
-    GCP_send_ooo (sender,
-                  env);
+    send_broken_without_mqm (sender,
+                             &msg->cid,
+                             &pids[off + 1]);
+    return;
+  }
+  if (max_routes <= GNUNET_CONTAINER_multishortmap_size (routes))
+  {
+    LOG (GNUNET_ERROR_TYPE_DEBUG,
+         "Received CADET_CONNECTION_CREATE from %s for %s. We have reached our route limit. Sending BROKEN\n",
+         GCP_2s (sender),
+         GNUNET_sh2s (&msg->cid.connection_of_tunnel));
+    send_broken_without_mqm (sender,
+                             &msg->cid,
+                             &pids[off - 1]);
     return;
   }
 
   /* Workable route, create routing entry */
   LOG (GNUNET_ERROR_TYPE_DEBUG,
-       "Received CONNECTION_CREATE from %s for %s. Next hop %s:%u is up. Creating route\n",
+       "Received CADET_CONNECTION_CREATE from %s for %s. Next hop %s:%u is up. Creating route\n",
        GCP_2s (sender),
        GNUNET_sh2s (&msg->cid.connection_of_tunnel),
        GNUNET_i2s (&pids[off + 1]),
        off + 1);
   route = GNUNET_new (struct CadetRoute);
   route->cid = msg->cid;
+  route->last_use = GNUNET_TIME_absolute_get ();
   dir_init (&route->prev,
             route,
             sender);
@@ -528,6 +652,14 @@ handle_connection_create (void *cls,
                                                      &route->cid.connection_of_tunnel,
                                                      route,
                                                      GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
+  route->hn = GNUNET_CONTAINER_heap_insert (route_heap,
+                                            route,
+                                            route->last_use.abs_value_us);
+  if (NULL == timeout_task)
+    timeout_task = GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_relative_multiply (keepalive_period,
+                                                                                3),
+                                                 &timeout_cb,
+                                                 NULL);
 }
 
 
@@ -608,7 +740,7 @@ handle_connection_broken (void *cls,
     LOG (GNUNET_ERROR_TYPE_DEBUG,
          "Received CONNECTION_BROKEN for connection %s. Destroying it.\n",
          GNUNET_sh2s (&msg->cid.connection_of_tunnel));
-    GCC_destroy (cc);
+    GCC_destroy_without_core (cc);
 
     /* FIXME: also destroy the path up to the specified link! */
     return;
@@ -658,7 +790,7 @@ handle_connection_destroy (void *cls,
          "Received CONNECTION_DESTROY for connection %s. Destroying connection.\n",
          GNUNET_sh2s (&msg->cid.connection_of_tunnel));
 
-    GCC_destroy (cc);
+    GCC_destroy_without_core (cc);
     return;
   }
 
@@ -715,6 +847,47 @@ handle_tunnel_kx (void *cls,
 }
 
 
+/**
+ * Handle for #GNUNET_MESSAGE_TYPE_CADET_TUNNEL_KX_AUTH
+ *
+ * @param cls Closure (CadetPeer for neighbor that sent the message).
+ * @param msg Message itself.
+ */
+static void
+handle_tunnel_kx_auth (void *cls,
+                       const struct GNUNET_CADET_TunnelKeyExchangeAuthMessage *msg)
+{
+  struct CadetPeer *peer = cls;
+  struct CadetConnection *cc;
+
+  /* First, check if message belongs to a connection that ends here. */
+  cc = GNUNET_CONTAINER_multishortmap_get (connections,
+                                           &msg->kx.cid.connection_of_tunnel);
+  if (NULL != cc)
+  {
+    /* verify message came from the right direction */
+    struct CadetPeerPath *path = GCC_get_path (cc);
+
+    if (peer !=
+        GCPP_get_peer_at_offset (path,
+                                 0))
+    {
+      /* received message from unexpected direction, ignore! */
+      GNUNET_break_op (0);
+      return;
+    }
+    GCC_handle_kx_auth (cc,
+                        msg);
+    return;
+  }
+
+  /* We're just an intermediary peer, route the message along its path */
+  route_message (peer,
+                 &msg->kx.cid,
+                 &msg->kx.header);
+}
+
+
 /**
  * Check if the encrypted message has the appropriate size.
  *
@@ -873,6 +1046,10 @@ GCO_init (const struct GNUNET_CONFIGURATION_Handle *c)
                              GNUNET_MESSAGE_TYPE_CADET_TUNNEL_KX,
                              struct GNUNET_CADET_TunnelKeyExchangeMessage,
                              NULL),
+    GNUNET_MQ_hd_fixed_size (tunnel_kx_auth,
+                             GNUNET_MESSAGE_TYPE_CADET_TUNNEL_KX_AUTH,
+                             struct GNUNET_CADET_TunnelKeyExchangeAuthMessage,
+                             NULL),
     GNUNET_MQ_hd_var_size (tunnel_encrypted,
                            GNUNET_MESSAGE_TYPE_CADET_TUNNEL_ENCRYPTED,
                            struct GNUNET_CADET_TunnelEncryptedMessage,
@@ -880,8 +1057,15 @@ GCO_init (const struct GNUNET_CONFIGURATION_Handle *c)
     GNUNET_MQ_handler_end ()
   };
 
+  if (GNUNET_OK !=
+      GNUNET_CONFIGURATION_get_value_number (c,
+                                             "CADET",
+                                             "MAX_ROUTES",
+                                             &max_routes))
+    max_routes = 10000;
   routes = GNUNET_CONTAINER_multishortmap_create (1024,
                                                   GNUNET_NO);
+  route_heap = GNUNET_CONTAINER_heap_create (GNUNET_CONTAINER_HEAP_ORDER_MIN);
   core = GNUNET_CORE_connect (c,
                               NULL,
                               &core_init_cb,
@@ -904,6 +1088,14 @@ GCO_shutdown ()
   }
   GNUNET_assert (0 == GNUNET_CONTAINER_multishortmap_size (routes));
   GNUNET_CONTAINER_multishortmap_destroy (routes);
+  routes = NULL;
+  GNUNET_CONTAINER_heap_destroy (route_heap);
+  route_heap = NULL;
+  if (NULL != timeout_task)
+  {
+    GNUNET_SCHEDULER_cancel (timeout_task);
+    timeout_task = NULL;
+  }
 }
 
 /* end of gnunet-cadet-service_core.c */