- fix queueing

6 files changed, 207 insertions, 182 deletions

diff --git a/src/mesh/gnunet-service-mesh_connection.c b/src/mesh/gnunet-service-mesh_connection.c
index 63e4b8bd0..e9e16293a 100644
--- a/src/mesh/gnunet-service-mesh_connection.c
+++ b/src/mesh/gnunet-service-mesh_connection.c
@@ -401,6 +401,27 @@ message_sent (void *cls,
     LOG (GNUNET_ERROR_TYPE_DEBUG, "!  destroying connection!\n");
     GMC_destroy (c);
   }
+  /* Send ACK if needed, after accounting for sent ID in fc->queue_n */
+  switch (type)
+  {
+    case GNUNET_MESSAGE_TYPE_MESH_FWD:
+    case GNUNET_MESSAGE_TYPE_MESH_BCK:
+      fc->last_pid_sent++;
+      LOG (GNUNET_ERROR_TYPE_DEBUG, "!   accounting pid %u\n", fc->last_pid_sent);
+//       send_ack (c, ch, fwd);
+      break;
+    default:
+      break;
+  }
+//   if (NULL != c->t)
+//   {
+//     c->t->pending_messages--;
+//     if (GNUNET_YES == c->t->destroy && 0 == t->pending_messages)
+//     {
+//       LOG (GNUNET_ERROR_TYPE_DEBUG, "*  destroying tunnel!\n");
+//       GMT_destroy (c->t);
+//     }
+//   }
 }
 
 
@@ -1776,6 +1797,20 @@ GMC_get_id (const struct MeshConnection *c)
 
 
 /**
+ * Get the connection path.
+ *
+ * @param c Connection to get the path from.
+ *
+ * @return path used by the connection.
+ */
+const struct MeshPeerPath *
+GMC_get_path (const struct MeshConnection *c)
+{
+  return c->path;
+}
+
+
+/**
  * Get the connection state.
  *
  * @param c Connection to get the state from.
diff --git a/src/mesh/gnunet-service-mesh_connection.h b/src/mesh/gnunet-service-mesh_connection.h
index 131bd9dc2..b0ddeabd9 100644
--- a/src/mesh/gnunet-service-mesh_connection.h
+++ b/src/mesh/gnunet-service-mesh_connection.h
@@ -264,6 +264,16 @@ const struct GNUNET_HashCode *
 GMC_get_id (const struct MeshConnection *c);
 
 /**
+ * Get the connection path.
+ *
+ * @param c Connection to get the path from.
+ *
+ * @return path used by the connection.
+ */
+const struct MeshPeerPath *
+GMC_get_path (const struct MeshConnection *c);
+
+/**
  * Get the connection state.
  *
  * @param c Connection to get the state from.
diff --git a/src/mesh/gnunet-service-mesh_peer.c b/src/mesh/gnunet-service-mesh_peer.c
index f997cf838..2bba99319 100644
--- a/src/mesh/gnunet-service-mesh_peer.c
+++ b/src/mesh/gnunet-service-mesh_peer.c
@@ -699,7 +699,6 @@ peer_get_best_path (const struct MeshPeer *peer)
 {
   struct MeshPeerPath *best_p;
   struct MeshPeerPath *p;
-  struct MeshConnection *c;
   unsigned int best_cost;
   unsigned int cost;
 
@@ -707,19 +706,56 @@ peer_get_best_path (const struct MeshPeer *peer)
   best_p = NULL;
   for (p = peer->path_head; NULL != p; p = p->next)
   {
-    for (c = peer->tunnel->connection_head; NULL != c; c = c->next)
-      if (c->path == p)
-        break;
-      if (NULL != c)
-        continue; /* If path is in use in a connection, skip it. */
-
-            if ((cost = peer_get_path_cost (peer, p)) < best_cost)
-            {
-              best_cost = cost;
-              best_p = p;
-            }
+    if (GNUNET_YES == GMT_is_path_used (peer->tunnel, p))
+      continue; /* If path is already in use, skip it. */
+
+    if ((cost = peer_get_path_cost (peer, p)) < best_cost)
+    {
+      best_cost = cost;
+      best_p = p;
+    }
+  }
+  return best_p;
+}
+
+
+static int
+queue_is_sendable (struct MeshPeerQueue *q)
+{
+  /* Is PID-independent? */
+  switch (q->type)
+  {
+    case GNUNET_MESSAGE_TYPE_MESH_ACK:
+    case GNUNET_MESSAGE_TYPE_MESH_POLL:
+      return GNUNET_YES;
   }
-    return best_p;
+
+  if (GMC_is_sendable (q->c, q->fwd))
+    return GNUNET_YES;
+
+  return GNUNET_NO;
+}
+
+
+/**
+ * Get first sendable message.
+ *
+ * @param peer The destination peer.
+ *
+ * @return Best current known path towards the peer, if any.
+ */
+static struct MeshPeerQueue *
+peer_get_first_message (const struct MeshPeer *peer)
+{
+  struct MeshPeerQueue *q;
+
+  for (q = peer->queue_head; NULL != q; q = q->next)
+  {
+    if (queue_is_sendable (q))
+      return q;
+  }
+
+  return NULL;
 }
 
 
@@ -737,16 +773,16 @@ search_handler (void *cls, const struct MeshPeerPath *path)
   struct MeshPeer *peer = cls;
   unsigned int connection_count;
 
-  path_add_to_peers (path, GNUNET_NO);
+  GMP_add_path_to_all (path, GNUNET_NO);
 
   /* Count connections */
-  connection_count = GMC_count (peer->tunnel->connection_head);
+  connection_count = GMT_count_connections (peer->tunnel);
 
   /* If we already have 3 (or more (?!)) connections, it's enough */
   if (3 <= connection_count)
     return;
 
-  if (peer->tunnel->state == MESH_TUNNEL3_SEARCHING)
+  if (MESH_TUNNEL3_SEARCHING == GMT_get_state (peer->tunnel))
   {
     LOG (GNUNET_ERROR_TYPE_DEBUG, " ... connect!\n");
     GMP_connect (peer);
@@ -756,6 +792,68 @@ search_handler (void *cls, const struct MeshPeerPath *path)
 
 
 /**
+ * Free a transmission that was already queued with all resources
+ * associated to the request.
+ *
+ * @param queue Queue handler to cancel.
+ * @param clear_cls Is it necessary to free associated cls?
+ */
+static void
+queue_destroy (struct MeshPeerQueue *queue, int clear_cls)
+{
+  struct MeshPeer *peer;
+
+  peer = queue->peer;
+  GNUNET_assert (NULL != queue->c);
+
+  if (GNUNET_YES == clear_cls)
+  {
+    LOG (GNUNET_ERROR_TYPE_DEBUG, "#   queue destroy type %s\n",
+                GNUNET_MESH_DEBUG_M2S (queue->type));
+    switch (queue->type)
+    {
+      case GNUNET_MESSAGE_TYPE_MESH_CONNECTION_DESTROY:
+      case GNUNET_MESSAGE_TYPE_MESH_TUNNEL_DESTROY:
+        LOG (GNUNET_ERROR_TYPE_INFO, "destroying a DESTROY message\n");
+        /* fall through */
+      case GNUNET_MESSAGE_TYPE_MESH_FWD:
+      case GNUNET_MESSAGE_TYPE_MESH_BCK:
+      case GNUNET_MESSAGE_TYPE_MESH_ACK:
+      case GNUNET_MESSAGE_TYPE_MESH_POLL:
+      case GNUNET_MESSAGE_TYPE_MESH_CONNECTION_ACK:
+      case GNUNET_MESSAGE_TYPE_MESH_CONNECTION_CREATE:
+      case GNUNET_MESSAGE_TYPE_MESH_CONNECTION_BROKEN:
+        LOG (GNUNET_ERROR_TYPE_DEBUG, "#   prebuilt message\n");;
+        GNUNET_free_non_null (queue->cls);
+        break;
+
+      default:
+        GNUNET_break (0);
+        LOG (GNUNET_ERROR_TYPE_ERROR, "#   type %s unknown!\n",
+                    GNUNET_MESH_DEBUG_M2S (queue->type));
+    }
+  }
+  GNUNET_CONTAINER_DLL_remove (peer->queue_head, peer->queue_tail, queue);
+
+  if (queue->type != GNUNET_MESSAGE_TYPE_MESH_ACK &&
+      queue->type != GNUNET_MESSAGE_TYPE_MESH_POLL)
+  {
+    peer->queue_n--;
+  }
+
+  if (NULL != queue->callback)
+  {
+    LOG (GNUNET_ERROR_TYPE_DEBUG, "#   Calling callback\n");
+    queue->callback (queue->callback_cls,
+                     queue->c, queue->type,
+                     queue->fwd, queue->size,
+                     GNUNET_TIME_absolute_get_duration (queue->start_waiting));
+  }
+
+  GNUNET_free (queue);
+}
+
+/**
  * Core callback to write a queued packet to core buffer
  *
  * @param cls Closure (peer info).
@@ -769,15 +867,9 @@ queue_send (void *cls, size_t size, void *buf)
 {
   struct MeshPeer *peer = cls;
   struct MeshConnection *c;
-  struct GNUNET_MessageHeader *msg;
   struct MeshPeerQueue *queue;
-  struct MeshTunnel3 *t;
-  struct MeshChannel *ch;
   const struct GNUNET_PeerIdentity *dst_id;
   size_t data_size;
-  uint32_t pid;
-  uint16_t type;
-  int fwd;
 
   peer->core_transmit = NULL;
   LOG (GNUNET_ERROR_TYPE_DEBUG, "* Queue send (max %u)\n", size);
@@ -796,7 +888,6 @@ queue_send (void *cls, size_t size, void *buf)
     return 0;
   }
   c = queue->c;
-  fwd = queue->fwd;
 
   dst_id = GNUNET_PEER_resolve2 (peer->id);
   LOG (GNUNET_ERROR_TYPE_DEBUG, "*   towards %s\n", GNUNET_i2s (dst_id));
@@ -817,9 +908,6 @@ queue_send (void *cls, size_t size, void *buf)
   }
   LOG (GNUNET_ERROR_TYPE_DEBUG, "*   size %u ok\n", queue->size);
 
-  t = (NULL != c) ? c->t : NULL;
-  type = 0;
-
   /* Fill buf */
   switch (queue->type)
   {
@@ -834,8 +922,6 @@ queue_send (void *cls, size_t size, void *buf)
                   "*   raw: %s\n",
                   GNUNET_MESH_DEBUG_M2S (queue->type));
       data_size = send_core_data_raw (queue->cls, size, buf);
-      msg = (struct GNUNET_MessageHeader *) buf;
-      type = ntohs (msg->type);
       break;
     case GNUNET_MESSAGE_TYPE_MESH_CONNECTION_CREATE:
       LOG (GNUNET_ERROR_TYPE_DEBUG, "*   path create\n");
@@ -876,29 +962,15 @@ queue_send (void *cls, size_t size, void *buf)
   }
 
   /* Free queue, but cls was freed by send_core_* */
-  ch = queue->ch;
-  GMP_queue_destroy (queue, GNUNET_NO);
-
-  /* Send ACK if needed, after accounting for sent ID in fc->queue_n */
-  switch (type)
-  {
-    case GNUNET_MESSAGE_TYPE_MESH_FWD:
-    case GNUNET_MESSAGE_TYPE_MESH_BCK:
-      pid = ntohl ( ((struct GNUNET_MESH_Encrypted *) buf)->pid );
-      LOG (GNUNET_ERROR_TYPE_DEBUG, "*   accounting pid %u\n", pid);
-      fc->last_pid_sent = pid;
-      send_ack (c, ch, fwd);
-      break;
-    default:
-      break;
-  }
+  queue_destroy (queue, GNUNET_NO);
 
   /* If more data in queue, send next */
   queue = peer_get_first_message (peer);
   if (NULL != queue)
   {
     LOG (GNUNET_ERROR_TYPE_DEBUG, "*   more data!\n");
-    if (NULL == peer->core_transmit) {
+    if (NULL == peer->core_transmit)
+    {
       peer->core_transmit =
           GNUNET_CORE_notify_transmit_ready(core_handle,
                                             0,
@@ -915,145 +987,22 @@ queue_send (void *cls, size_t size, void *buf)
       LOG (GNUNET_ERROR_TYPE_DEBUG,
                   "*   tmt rdy called somewhere else\n");
     }
-    if (GNUNET_SCHEDULER_NO_TASK == fc->poll_task)
-    {
-      LOG (GNUNET_ERROR_TYPE_DEBUG, "*   starting poll timeout\n");
-      fc->poll_task =
-          GNUNET_SCHEDULER_add_delayed (fc->poll_time, &connection_poll, fc);
-    }
+//     GMC_start_poll (); FIXME needed?
   }
   else
   {
-    if (GNUNET_SCHEDULER_NO_TASK != fc->poll_task)
-    {
-      GNUNET_SCHEDULER_cancel (fc->poll_task);
-      fc->poll_task = GNUNET_SCHEDULER_NO_TASK;
-    }
+//     GMC_stop_poll(); FIXME needed?
   }
 
-  if (NULL != t)
-  {
-    t->pending_messages--;
-    if (GNUNET_YES == t->destroy && 0 == t->pending_messages)
-    {
-      LOG (GNUNET_ERROR_TYPE_DEBUG, "*  destroying tunnel!\n");
-      GMT_destroy (t);
-    }
-  }
   LOG (GNUNET_ERROR_TYPE_DEBUG, "*  Return %d\n", data_size);
   return data_size;
 }
 
 
-static int
-queue_is_sendable (struct MeshPeerQueue *q)
-{
-  /* Is PID-independent? */
-  switch (q->type)
-  {
-    case GNUNET_MESSAGE_TYPE_MESH_ACK:
-    case GNUNET_MESSAGE_TYPE_MESH_POLL:
-      return GNUNET_YES;
-  }
-
-  if (GMC_is_sendable (q->c, q->fwd))
-    return GNUNET_YES;
-
-  return GNUNET_NO;
-}
-
-/**
- * Get first sendable message.
- *
- * @param peer The destination peer.
- *
- * @return Best current known path towards the peer, if any.
- */
-static struct MeshPeerQueue *
-peer_get_first_message (const struct MeshPeer *peer)
-{
-  struct MeshPeerQueue *q;
-
-  for (q = peer->queue_head; NULL != q; q = q->next)
-  {
-    if (queue_is_sendable (q))
-      return q;
-  }
-
-  return NULL;
-}
-
-
-
-
 /******************************************************************************/
 /********************************    API    ***********************************/
 /******************************************************************************/
 
-
-/**
- * Free a transmission that was already queued with all resources
- * associated to the request.
- *
- * @param queue Queue handler to cancel.
- * @param clear_cls Is it necessary to free associated cls?
- */
-void
-GMP_queue_destroy (struct MeshPeerQueue *queue, int clear_cls)
-{
-  struct MeshPeer *peer;
-
-  peer = queue->peer;
-  GNUNET_assert (NULL != queue->c);
-
-  if (GNUNET_YES == clear_cls)
-  {
-    LOG (GNUNET_ERROR_TYPE_DEBUG, "#   queue destroy type %s\n",
-                GNUNET_MESH_DEBUG_M2S (queue->type));
-    switch (queue->type)
-    {
-      case GNUNET_MESSAGE_TYPE_MESH_CONNECTION_DESTROY:
-      case GNUNET_MESSAGE_TYPE_MESH_TUNNEL_DESTROY:
-        LOG (GNUNET_ERROR_TYPE_INFO, "destroying a DESTROY message\n");
-        /* fall through */
-      case GNUNET_MESSAGE_TYPE_MESH_FWD:
-      case GNUNET_MESSAGE_TYPE_MESH_BCK:
-      case GNUNET_MESSAGE_TYPE_MESH_ACK:
-      case GNUNET_MESSAGE_TYPE_MESH_POLL:
-      case GNUNET_MESSAGE_TYPE_MESH_CONNECTION_ACK:
-      case GNUNET_MESSAGE_TYPE_MESH_CONNECTION_CREATE:
-      case GNUNET_MESSAGE_TYPE_MESH_CONNECTION_BROKEN:
-        LOG (GNUNET_ERROR_TYPE_DEBUG, "#   prebuilt message\n");;
-        GNUNET_free_non_null (queue->cls);
-        break;
-
-      default:
-        GNUNET_break (0);
-        LOG (GNUNET_ERROR_TYPE_ERROR, "#   type %s unknown!\n",
-                    GNUNET_MESH_DEBUG_M2S (queue->type));
-    }
-  }
-  GNUNET_CONTAINER_DLL_remove (peer->queue_head, peer->queue_tail, queue);
-
-  if (queue->type != GNUNET_MESSAGE_TYPE_MESH_ACK &&
-      queue->type != GNUNET_MESSAGE_TYPE_MESH_POLL)
-  {
-    peer->queue_n--;
-  }
-
-  if (NULL != queue->callback)
-  {
-    LOG (GNUNET_ERROR_TYPE_DEBUG, "#   Calling callback\n");
-    queue->callback (queue->callback_cls,
-                     queue->c, queue->type,
-                     queue->fwd, queue->size,
-                     GNUNET_TIME_absolute_get_duration (queue->start_waiting));
-  }
-
-  GNUNET_free (queue);
-}
-
-
 /**
  * @brief Queue and pass message to core when possible.
  *
@@ -1121,7 +1070,7 @@ GMP_queue_add (struct MeshPeer *peer, void *cls, uint16_t type, size_t size,
       if (copy->type == type && copy->c == c && copy->fwd == fwd)
       {
         /* Example: also a FWD ACK for connection XYZ */
-        GMP_queue_destroy (copy, GNUNET_YES);
+        queue_destroy (copy, GNUNET_YES);
       }
     }
     GNUNET_CONTAINER_DLL_insert (peer->queue_head, peer->queue_tail, queue);
@@ -1136,7 +1085,7 @@ GMP_queue_add (struct MeshPeer *peer, void *cls, uint16_t type, size_t size,
   {
     LOG (GNUNET_ERROR_TYPE_DEBUG,
                 "calling core tmt rdy towards %s for %u bytes\n",
-                peer2s (peer), size);
+                GMP_2s (peer), size);
     peer->core_transmit =
         GNUNET_CORE_notify_transmit_ready (core_handle,
                                            0,
@@ -1152,7 +1101,7 @@ GMP_queue_add (struct MeshPeer *peer, void *cls, uint16_t type, size_t size,
   {
     LOG (GNUNET_ERROR_TYPE_DEBUG,
                 "core tmt rdy towards %s already called\n",
-                peer2s (peer));
+                GMP_2s (peer));
 
   }
 }
@@ -1178,7 +1127,7 @@ GMP_queue_cancel (struct MeshPeer *peer, struct MeshConnection *c)
       LOG (GNUNET_ERROR_TYPE_DEBUG,
                   "connection_cancel_queue %s\n",
                   GNUNET_MESH_DEBUG_M2S (q->type));
-      GMP_queue_destroy (q, GNUNET_YES);
+      queue_destroy (q, GNUNET_YES);
     }
   }
   if (NULL == peer->queue_head)
@@ -1380,7 +1329,7 @@ GMP_connect (struct MeshPeer *peer)
 
   LOG (GNUNET_ERROR_TYPE_DEBUG,
               "peer_connect towards %s\n",
-              peer2s (peer));
+              GMP_2s (peer));
   t = peer->tunnel;
   c = NULL;
   rerun_search = GNUNET_NO;
@@ -1433,7 +1382,7 @@ GMP_connect (struct MeshPeer *peer)
 
     id = GNUNET_PEER_resolve2 (peer->id);
     LOG (GNUNET_ERROR_TYPE_DEBUG,
-                "  Starting DHT GET for peer %s\n", peer2s (peer));
+                "  Starting DHT GET for peer %s\n", GMP_2s (peer));
     peer->search_h = GMD_search (id, &search_handler, peer);
     if (MESH_TUNNEL3_NEW == GMT_get_state (t))
       GMT_change_state (t, MESH_TUNNEL3_SEARCHING);
@@ -1576,7 +1525,7 @@ GMP_add_path (struct MeshPeer *peer_info, struct MeshPeerPath *path,
   }
 
   LOG (GNUNET_ERROR_TYPE_DEBUG, "adding path [%u] to peer %s\n",
-              path->length, peer2s (peer_info));
+              path->length, GMP_2s (peer_info));
 
   l = path_get_length (path);
   if (0 == l)
@@ -1640,7 +1589,7 @@ GMP_add_path_to_origin (struct MeshPeer *peer,
  * @param confirmed Whether we know if the path works or not.
  */
 void
-GMP_add_path_to_all (struct MeshPeerPath *p, int confirmed)
+GMP_add_path_to_all (const struct MeshPeerPath *p, int confirmed)
 {
   unsigned int i;
 
diff --git a/src/mesh/gnunet-service-mesh_peer.h b/src/mesh/gnunet-service-mesh_peer.h
index c7265ce2f..57eb1c61d 100644
--- a/src/mesh/gnunet-service-mesh_peer.h
+++ b/src/mesh/gnunet-service-mesh_peer.h
@@ -45,6 +45,8 @@ extern "C"
  */
 struct MeshPeer;
 
+#include "gnunet-service-mesh_connection.h"
+
 /**
  * Callback called when a queued message is sent.
  *
@@ -60,9 +62,6 @@ typedef void (*GMP_sent) (void *cls,
                           int fwd, size_t size,
                           struct GNUNET_TIME_Relative wait);
 
-#include "gnunet-service-mesh_connection.h"
-
-
 /******************************************************************************/
 /********************************    API    ***********************************/
 /******************************************************************************/
@@ -225,7 +224,7 @@ GMP_add_path_to_origin (struct MeshPeer *peer_info,
  * @param confirmed Whether we know if the path works or not.
  */
 void
-GMP_add_path_to_all (struct MeshPeerPath *p, int confirmed);
+GMP_add_path_to_all (const struct MeshPeerPath *p, int confirmed);
 
 /**
  * Remove a connection from a neighboring peer.
diff --git a/src/mesh/gnunet-service-mesh_tunnel.c b/src/mesh/gnunet-service-mesh_tunnel.c
index c5ccc9f5c..9341e0eaf 100644
--- a/src/mesh/gnunet-service-mesh_tunnel.c
+++ b/src/mesh/gnunet-service-mesh_tunnel.c
@@ -1170,6 +1170,27 @@ GMT_is_loopback (const struct MeshTunnel3 *t)
 
 
 /**
+ * Is the tunnel using this path already?
+ *
+ * @param t Tunnel.
+ * @param p Path.
+ *
+ * @return GNUNET_YES a connection uses this path.
+ */
+int
+GMT_is_path_used (const struct MeshTunnel3 *t, const struct MeshPeerPath *p)
+{
+  struct MeshTConnection *c;
+
+  for (c = t->connection_head; NULL != c; c = c->next)
+    if (GMC_get_path (c) == p)
+      return GNUNET_YES;
+
+  return GNUNET_NO;
+}
+
+
+/**
  * Get the static string for the peer this tunnel is directed.
  *
  * @param t Tunnel.
diff --git a/src/mesh/gnunet-service-mesh_tunnel.h b/src/mesh/gnunet-service-mesh_tunnel.h
index c55e91fe2..89c8a1462 100644
--- a/src/mesh/gnunet-service-mesh_tunnel.h
+++ b/src/mesh/gnunet-service-mesh_tunnel.h
@@ -321,6 +321,17 @@ int
 GMT_is_loopback (const struct MeshTunnel3 *t);
 
 /**
+ * Is the tunnel using this path already?
+ *
+ * @param t Tunnel.
+ * @param p Path.
+ *
+ * @return GNUNET_YES a connection uses this path.
+ */
+int
+GMT_is_path_used (const struct MeshTunnel3 *t, const struct MeshPeerPath *p);
+
+/**
  * Get the static string for the peer this tunnel is directed.
  *
  * @param t Tunnel.