- use blocks for checking by type results
[oweals/gnunet.git] / src / mesh / gnunet-service-mesh_new.c
index 7ff9f5e99752fefe701f801c87f586b5356238a6..932fbb3b5a0149777077495b0a2058803ee02ee0 100644 (file)
@@ -48,6 +48,7 @@
 #include "platform.h"
 #include "mesh.h"
 #include "mesh_protocol.h"
+#include "block_mesh.h"
 #include "gnunet_dht_service.h"
 #include "mesh_tunnel_tree.h"
 
 
 /* TODO END */
 
-#define MESH_DEBUG_DHT GNUNET_YES
-#define MESH_DEBUG_CONNECTION GNUNET_NO
+#define MESH_BLOOM_SIZE         128
+
+#define MESH_DEBUG_DHT          GNUNET_YES
+#define MESH_DEBUG_CONNECTION   GNUNET_NO
 
 #if MESH_DEBUG_CONNECTION
 #define DEBUG_CONN(...) GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, __VA_ARGS__)
@@ -136,6 +139,11 @@ struct MeshPeerQueue
      */
   struct MeshPeerInfo *peer;
 
+    /**
+     * Tunnel this message belongs to.
+     */
+  struct MeshTunnel *tunnel;
+
     /**
      * Pointer to info stucture used as cls.
      */
@@ -225,46 +233,29 @@ struct MeshPeerInfo
      * Number of tunnels this peers participates in
      */
   unsigned int ntunnels;
-};
-
-
-/**
- * Data scheduled to transmit (to local client or remote peer)
- */
-struct MeshQueue
-{
-    /**
-     * Double linked list
-     */
-  struct MeshQueue *next;
-  struct MeshQueue *prev;
-
-    /**
-     * Target of the data (NULL if target is client)
-     */
-  struct MeshPeerInfo *peer;
 
-    /**
-     * Client to send the data to (NULL if target is peer)
-     */
-  struct MeshClient *client;
+   /**
+    * Transmission queue to core DLL head
+    */
+  struct MeshPeerQueue *queue_head;
 
-    /**
-     * Size of the message to transmit
-     */
-  unsigned int size;
+   /**
+    * Transmission queue to core DLL tail
+    */
+   struct MeshPeerQueue *queue_tail;
 
-    /**
-     * How old is the data?
-     */
-  struct GNUNET_TIME_Absolute timestamp;
+   /**
+    * How many messages are in the queue to this peer.
+    */
+   unsigned int queue_n;
 
-    /**
-     * Data itself
-     */
-  struct GNUNET_MessageHeader *data;
+   /**
+    * Handle to for queued transmissions
+    */
+  struct GNUNET_CORE_TransmitHandle *core_transmit;
 };
 
+
 /**
  * Globally unique tunnel identification (owner + number)
  * DO NOT USE OVER THE NETWORK
@@ -313,10 +304,30 @@ struct MeshTunnel
   MESH_TunnelNumber local_tid_dest;
 
     /**
-     * ID of the last multicast packet seen/sent.
+     * Global count ID of the last *multicast* packet seen/sent.
      */
   uint32_t mid;
 
+    /**
+     * Local count ID of the last packet seen/sent.
+     */
+  uint32_t pid;
+
+    /**
+     * SKIP value for this tunnel.
+     */
+  uint32_t skip;
+
+    /**
+     * How many messages are in the queue.
+     */
+   unsigned int queue_n;
+
+    /**
+     * How many messages do we accept in the queue.
+     */
+   unsigned int queue_max;
+
     /**
      * Last time the tunnel was used
      */
@@ -364,10 +375,19 @@ struct MeshTunnel
   unsigned int nignore;
 
     /**
-     * Messages ready to transmit
+     * Blacklisted peers
+     */
+  GNUNET_PEER_Id *blacklisted;
+
+    /**
+     * Number of elements in blacklisted
      */
-  struct MeshQueue *queue_head;
-  struct MeshQueue *queue_tail;
+  unsigned int nblacklisted;
+
+  /**
+   * Bloomfilter (for peer identities) to stop circular routes
+   */
+  char bloomfilter[MESH_BLOOM_SIZE];
 
   /**
    * Tunnel paths
@@ -517,12 +537,6 @@ mesh_debug (void *cls, int success)
 static struct MeshClient *clients;
 static struct MeshClient *clients_tail;
 
-/**
- * Transmission queue to core
- */
-struct MeshPeerQueue *queue_head;
-struct MeshPeerQueue *queue_tail;
-
 /**
  * Tunnels known, indexed by MESH_TunnelID (MeshTunnel)
  */
@@ -549,11 +563,6 @@ static struct GNUNET_CONTAINER_MultiHashMap *peers;
  */
 static struct GNUNET_CORE_Handle *core_handle;
 
-/**
- * Handle to for queued transmissions
- */
-struct GNUNET_CORE_TransmitHandle *core_transmit;
-
 /**
  * Handle to use DHT
  */
@@ -648,15 +657,28 @@ unsigned int next_client_id;
  *         GNUNET_NO if not.
  */
 static int
-announce_application (void *cls, const GNUNET_HashCode * key, void *value)
+announce_application (void *cls, const struct GNUNET_HashCode * key, void *value)
 {
+  struct PBlock block;
+  struct MeshClient *c;
+
+  block.id = my_full_id;
+  c =  GNUNET_CONTAINER_multihashmap_get (applications, key);
+  block.type = (long) GNUNET_CONTAINER_multihashmap_get (c->apps, key);
+  if (0 == block.type)
+  {
+    GNUNET_break(0);
+    return GNUNET_YES;
+  }
+  block.type = htonl (block.type);
   /* FIXME are hashes in multihash map equal on all aquitectures? */
   /* FIXME: keep return value of 'put' to possibly cancel!? */
   GNUNET_DHT_put (dht_handle, key, 10,
                   GNUNET_DHT_RO_RECORD_ROUTE |
-                  GNUNET_DHT_RO_DEMULTIPLEX_EVERYWHERE, GNUNET_BLOCK_TYPE_TEST,
-                  sizeof (struct GNUNET_PeerIdentity),
-                  (const char *) &my_full_id,
+                  GNUNET_DHT_RO_DEMULTIPLEX_EVERYWHERE,
+                  GNUNET_BLOCK_TYPE_MESH_PEER_BY_TYPE,
+                  sizeof (block),
+                  (const char *) &block,
                   GNUNET_TIME_absolute_add (GNUNET_TIME_absolute_get (),
                                             APP_ANNOUNCE_TIME),
                   APP_ANNOUNCE_TIME, NULL, NULL);
@@ -701,6 +723,8 @@ announce_applications (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
 static void
 announce_id (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
 {
+  struct PBlock block;
+
   if (tc->reason == GNUNET_SCHEDULER_REASON_SHUTDOWN)
   {
     announce_id_task = GNUNET_SCHEDULER_NO_TASK;
@@ -712,13 +736,15 @@ announce_id (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
    */
   DEBUG_DHT ("DHT_put for ID %s started.\n", GNUNET_i2s (&my_full_id));
 
+  block.id = my_full_id;
+  block.type = htonl (0);
   GNUNET_DHT_put (dht_handle,   /* DHT handle */
                   &my_full_id.hashPubKey,       /* Key to use */
                   10,          /* Replication level */
                   GNUNET_DHT_RO_RECORD_ROUTE | GNUNET_DHT_RO_DEMULTIPLEX_EVERYWHERE,    /* DHT options */
-                  GNUNET_BLOCK_TYPE_TEST,       /* Block type */
-                  sizeof (my_full_id),  /* Size of the data */
-                  (char *) &my_full_id, /* Data itself */
+                  GNUNET_BLOCK_TYPE_MESH_PEER,       /* Block type */
+                  sizeof (block),  /* Size of the data */
+                  (const char *) &block, /* Data itself */
                   GNUNET_TIME_UNIT_FOREVER_ABS,  /* Data expiration */
                   GNUNET_TIME_UNIT_FOREVER_REL, /* Retry time */
                   NULL,         /* Continuation */
@@ -742,7 +768,7 @@ announce_id (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
  */
 static void
 dht_get_id_handler (void *cls, struct GNUNET_TIME_Absolute exp,
-                    const GNUNET_HashCode * key,
+                    const struct GNUNET_HashCode * key,
                     const struct GNUNET_PeerIdentity *get_path,
                     unsigned int get_path_length,
                     const struct GNUNET_PeerIdentity *put_path,
@@ -827,7 +853,7 @@ client_get (struct GNUNET_SERVER_Client *client)
 static int
 client_is_subscribed (uint16_t message_type, struct MeshClient *c)
 {
-  GNUNET_HashCode hc;
+  struct GNUNET_HashCode hc;
 
   GNUNET_CRYPTO_hash (&message_type, sizeof (uint16_t), &hc);
   return GNUNET_CONTAINER_multihashmap_contains (c->types, &hc);
@@ -916,7 +942,7 @@ client_knows_tunnel (struct MeshClient *c, struct MeshTunnel *t)
 static void
 client_ignore_tunnel (struct MeshClient *c, struct MeshTunnel *t)
 {
-  GNUNET_HashCode hash;
+  struct GNUNET_HashCode hash;
 
   GNUNET_CRYPTO_hash(&t->local_tid_dest, sizeof (MESH_TunnelNumber), &hash);
   GNUNET_break (GNUNET_YES ==
@@ -940,7 +966,7 @@ client_ignore_tunnel (struct MeshClient *c, struct MeshTunnel *t)
 static void
 client_delete_tunnel (struct MeshClient *c, struct MeshTunnel *t)
 {
-  GNUNET_HashCode hash;
+  struct GNUNET_HashCode hash;
 
   if (c == t->owner)
   {
@@ -1040,7 +1066,7 @@ send_subscribed_clients (const struct GNUNET_MessageHeader *msg,
         {
           /* This client doesn't know the tunnel */
           struct GNUNET_MESH_TunnelNotification tmsg;
-          GNUNET_HashCode hash;
+          struct GNUNET_HashCode hash;
 
           tmsg.header.size = htons (sizeof (tmsg));
           tmsg.header.type = htons (GNUNET_MESSAGE_TYPE_MESH_LOCAL_TUNNEL_CREATE);
@@ -1241,7 +1267,7 @@ peer_info_get_short (const GNUNET_PEER_Id peer)
  * @return always GNUNET_YES, to keep iterating
  */
 static int
-peer_info_delete_tunnel (void *cls, const GNUNET_HashCode * key, void *value)
+peer_info_delete_tunnel (void *cls, const struct GNUNET_HashCode * key, void *value)
 {
   struct MeshTunnel *t = cls;
   struct MeshPeerInfo *peer = value;
@@ -1269,9 +1295,11 @@ peer_info_delete_tunnel (void *cls, const GNUNET_HashCode * key, void *value)
  * @param type Type of the message.
  * @param size Size of the message.
  * @param dst Neighbor to send message to.
+ * @param t Tunnel this message belongs to.
  */
 static void
-queue_add (void *cls, uint16_t type, size_t size, struct MeshPeerInfo *dst);
+queue_add (void *cls, uint16_t type, size_t size,
+           struct MeshPeerInfo *dst, struct MeshTunnel *t);
 
 /**
   * Core callback to write a pre-constructed data packet to core buffer
@@ -1307,17 +1335,17 @@ send_core_data_raw (void *cls, size_t size, void *buf)
 
 
 /**
- * Sends an already built message to a peer, properly registrating
+ * Sends an already built unicast message to a peer, properly registrating
  * all used resources.
  *
- * @param message Message to send. Fucntion makes a copy of it.
+ * @param message Message to send. Function makes a copy of it.
  * @param peer Short ID of the neighbor whom to send the message.
- *
- * FIXME tunnel?
+ * @param t Tunnel on which this message is transmitted.
  */
 static void
 send_message (const struct GNUNET_MessageHeader *message,
-              const struct GNUNET_PeerIdentity *peer)
+              const struct GNUNET_PeerIdentity *peer,
+              struct MeshTunnel *t)
 {
   struct MeshTransmissionDescriptor *info;
   struct MeshPeerInfo *neighbor;
@@ -1352,7 +1380,8 @@ send_message (const struct GNUNET_MessageHeader *message,
   queue_add (info,
              GNUNET_MESSAGE_TYPE_MESH_UNICAST,
              size,
-             neighbor);
+             neighbor,
+             t);
 }
 
 
@@ -1405,7 +1434,8 @@ send_create_path (struct MeshPeerInfo *peer, struct MeshPeerPath *p,
              GNUNET_MESSAGE_TYPE_MESH_PATH_CREATE,
              sizeof (struct GNUNET_MESH_ManipulatePath) +
                 (p->length * sizeof (struct GNUNET_PeerIdentity)),
-             neighbor);
+             neighbor,
+             t);
 }
 
 
@@ -1444,7 +1474,7 @@ send_destroy_path (struct MeshTunnel *t, GNUNET_PEER_Id destination)
     {
       GNUNET_PEER_resolve (p->peers[i], &pi[i]);
     }
-    send_message (&msg->header, tree_get_first_hop (t->tree, destination));
+    send_message (&msg->header, tree_get_first_hop (t->tree, destination), t);
   }
   path_destroy (p);
 }
@@ -1481,7 +1511,7 @@ peer_info_connect (struct MeshPeerInfo *peer, struct MeshTunnel *t)
     }
     else
     {
-      GNUNET_HashCode hash;
+      struct GNUNET_HashCode hash;
 
       path_destroy (p);
       send_client_peer_connected (t, myid);
@@ -1509,11 +1539,13 @@ peer_info_connect (struct MeshPeerInfo *peer, struct MeshTunnel *t)
                 "  Starting DHT GET for peer %s\n", GNUNET_i2s (&id));
     peer->dhtgetcls = path_info;
     peer->dhtget = GNUNET_DHT_get_start (dht_handle,    /* handle */
-                                         GNUNET_BLOCK_TYPE_TEST,        /* type */
-                                         &id.hashPubKey,        /* key to search */
+                                         GNUNET_BLOCK_TYPE_MESH_PEER, /* type */
+                                         &id.hashPubKey,     /* key to search */
                                          10,     /* replication level */
-                                         GNUNET_DHT_RO_RECORD_ROUTE | GNUNET_DHT_RO_DEMULTIPLEX_EVERYWHERE, NULL,       /* xquery */
-                                         0,     /* xquery bits */
+                                         GNUNET_DHT_RO_RECORD_ROUTE |
+                                         GNUNET_DHT_RO_DEMULTIPLEX_EVERYWHERE,
+                                         NULL,       /* xquery */ // FIXME BLOOMFILTER
+                                         0,     /* xquery bits */ // FIXME BLOOMFILTER SIZE
                                          &dht_get_id_handler, path_info);
   }
   /* Otherwise, there is no path but the DHT get is already started. */
@@ -1918,7 +1950,7 @@ path_refresh (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc);
 static struct MeshTunnel *
 tunnel_get_incoming (MESH_TunnelNumber tid)
 {
-  GNUNET_HashCode hash;
+  struct GNUNET_HashCode hash;
 
   GNUNET_assert (tid >= GNUNET_MESH_LOCAL_TUNNEL_ID_SERV);
   GNUNET_CRYPTO_hash (&tid, sizeof (MESH_TunnelNumber), &hash);
@@ -1943,7 +1975,7 @@ tunnel_get_by_local_id (struct MeshClient *c, MESH_TunnelNumber tid)
   }
   else
   {
-    GNUNET_HashCode hash;
+    struct GNUNET_HashCode hash;
 
     GNUNET_CRYPTO_hash (&tid, sizeof (MESH_TunnelNumber), &hash);
     return GNUNET_CONTAINER_multihashmap_get (c->own_tunnels, &hash);
@@ -1963,7 +1995,7 @@ static struct MeshTunnel *
 tunnel_get_by_pi (GNUNET_PEER_Id pi, MESH_TunnelNumber tid)
 {
   struct MESH_TunnelID id;
-  GNUNET_HashCode hash;
+  struct GNUNET_HashCode hash;
 
   id.oid = pi;
   id.tid = tid;
@@ -2204,7 +2236,7 @@ tunnel_notify_connection_broken (struct MeshTunnel *t, GNUNET_PEER_Id p1,
       msg.peer1 = my_full_id;
       GNUNET_PEER_resolve (pid, &msg.peer2);
       GNUNET_PEER_resolve (tree_get_predecessor (t->tree), &neighbor);
-      send_message (&msg.header, &neighbor);
+      send_message (&msg.header, &neighbor, t);
     }
   }
   return pid;
@@ -2237,8 +2269,8 @@ tunnel_send_multicast_iterator (void *cls, GNUNET_PEER_Id neighbor_id)
   queue_add(info,
             GNUNET_MESSAGE_TYPE_MESH_MULTICAST,
             info->mesh_data->data_len,
-            info->peer
-           );
+            info->peer,
+            mdata->t);
 }
 
 /**
@@ -2339,9 +2371,7 @@ static int
 tunnel_destroy (struct MeshTunnel *t)
 {
   struct MeshClient *c;
-  struct MeshQueue *q;
-  struct MeshQueue *qn;
-  GNUNET_HashCode hash;
+  struct GNUNET_HashCode hash;
   unsigned int i;
   int r;
 
@@ -2409,16 +2439,7 @@ tunnel_destroy (struct MeshTunnel *t)
                                            t);
     GNUNET_CONTAINER_multihashmap_destroy (t->peers);
   }
-  q = t->queue_head;
-  while (NULL != q)
-  {
-    if (NULL != q->data)
-      GNUNET_free (q->data);
-    qn = q->next;
-    GNUNET_free (q);
-    q = qn;
-    /* TODO cancel core transmit ready in case it was active */
-  }
+
   tree_destroy (t->tree);
   if (NULL != t->dht_get_type)
     GNUNET_DHT_get_stop (t->dht_get_type);
@@ -2431,6 +2452,62 @@ tunnel_destroy (struct MeshTunnel *t)
 }
 
 
+/**
+ * Create a new tunnel
+ * 
+ * @param owner Who is the owner of the tunnel (short ID).
+ * @param tid Tunnel Number of the tunnel.
+ * @param client Clients that owns the tunnel, NULL for foreign tunnels.
+ * @param local Tunnel Number for the tunnel, for the client point of view.
+ * 
+ */
+static struct MeshTunnel *
+tunnel_new (GNUNET_PEER_Id owner,
+            MESH_TunnelNumber tid,
+            struct MeshClient *client,
+            MESH_TunnelNumber local)
+{
+  struct MeshTunnel *t;
+  struct GNUNET_HashCode hash;
+
+  t = GNUNET_malloc (sizeof (struct MeshTunnel));
+  t->id.oid = owner;
+  t->id.tid = tid;
+  t->queue_max = 1000; // FIXME API parameter
+  t->tree = tree_new (owner);
+  t->owner = client;
+  t->local_tid = local;
+
+  GNUNET_CRYPTO_hash (&t->id, sizeof (struct MESH_TunnelID), &hash);
+  if (GNUNET_OK !=
+      GNUNET_CONTAINER_multihashmap_put (tunnels, &hash, t,
+                                         GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY))
+  {
+    GNUNET_break (0);
+    tunnel_destroy (t);
+    if (NULL != client)
+      GNUNET_SERVER_receive_done (client->handle, GNUNET_SYSERR);
+    return NULL;
+  }
+
+  if (NULL != client)
+  {
+    GNUNET_CRYPTO_hash (&t->local_tid, sizeof (MESH_TunnelNumber), &hash);
+    if (GNUNET_OK !=
+        GNUNET_CONTAINER_multihashmap_put (client->own_tunnels, &hash, t,
+                                          GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY))
+    {
+      GNUNET_break (0);
+      tunnel_destroy (t);
+      GNUNET_SERVER_receive_done (client->handle, GNUNET_SYSERR);
+      return NULL;
+    }
+  }
+
+  return t;
+}
+
+
 /**
  * Removes an explicit path from a tunnel, freeing all intermediate nodes
  * that are no longer needed, as well as nodes of no longer reachable peers.
@@ -2460,7 +2537,7 @@ tunnel_delete_peer (struct MeshTunnel *t, GNUNET_PEER_Id peer)
  * @return GNUNET_OK on success
  */
 static int
-tunnel_destroy_iterator (void *cls, const GNUNET_HashCode * key, void *value)
+tunnel_destroy_iterator (void *cls, const struct GNUNET_HashCode * key, void *value)
 {
   struct MeshTunnel *t = value;
   struct MeshClient *c = cls;
@@ -2677,7 +2754,6 @@ queue_destroy (struct MeshPeerQueue *queue, int clear_cls)
     switch (queue->type)
     {
     case GNUNET_MESSAGE_TYPE_MESH_UNICAST:
-    case GNUNET_MESSAGE_TYPE_MESH_TO_ORIGIN:
     case GNUNET_MESSAGE_TYPE_MESH_MULTICAST:
         GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "   type payload\n");
         dd = queue->cls;
@@ -2694,7 +2770,9 @@ queue_destroy (struct MeshPeerQueue *queue, int clear_cls)
     }
     GNUNET_free_non_null (queue->cls);
   }
-  GNUNET_CONTAINER_DLL_remove (queue_head, queue_tail, queue);
+  GNUNET_CONTAINER_DLL_remove (queue->peer->queue_head,
+                               queue->peer->queue_tail,
+                               queue);
   GNUNET_free (queue);
 }
 
@@ -2715,8 +2793,11 @@ queue_send (void *cls, size_t size, void *buf)
     struct MeshPeerQueue *queue;
     size_t data_size;
 
-    core_transmit = NULL;
-    queue = queue_head;
+    peer->core_transmit = NULL;
+    queue = peer->queue_head;
+
+    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "********* Queue send\n");
+
 
     /* If queue is empty, send should have been cancelled */
     if (NULL == queue)
@@ -2724,14 +2805,15 @@ queue_send (void *cls, size_t size, void *buf)
         GNUNET_break(0);
         return 0;
     }
+    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "*********   not empty\n");
 
     /* Check if buffer size is enough for the message */
-    if (queue->size < size)
+    if (queue->size > size)
     {
         struct GNUNET_PeerIdentity id;
 
         GNUNET_PEER_resolve (peer->id, &id);
-        core_transmit =
+        peer->core_transmit =
             GNUNET_CORE_notify_transmit_ready(core_handle,
                                               0,
                                               0,
@@ -2742,50 +2824,55 @@ queue_send (void *cls, size_t size, void *buf)
                                               peer);
         return 0;
     }
+    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "*********   size ok\n");
 
     /* Fill buf */
     switch (queue->type)
     {
         case GNUNET_MESSAGE_TYPE_MESH_UNICAST:
+            GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "*********   unicast\n");
             data_size = send_core_data_raw (queue->cls, size, buf);
             break;
         case GNUNET_MESSAGE_TYPE_MESH_MULTICAST:
+            GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "*********   multicast\n");
             data_size = send_core_data_multicast(queue->cls, size, buf);
             break;
-        case GNUNET_MESSAGE_TYPE_MESH_TO_ORIGIN:
-            data_size = 0;
-            break;
         case GNUNET_MESSAGE_TYPE_MESH_PATH_CREATE:
+            GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "*********   path create\n");
             data_size = send_core_path_create(queue->cls, size, buf);
             break;
         case GNUNET_MESSAGE_TYPE_MESH_PATH_ACK:
+            GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "*********   path ack\n");
             data_size = send_core_path_ack(queue->cls, size, buf);
             break;
         default:
             GNUNET_break (0);
-            GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "   type unknown!\n");
+            GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "*********   type unknown\n");
             data_size = 0;
     }
+    queue->tunnel->queue_n--;
 
     /* Free queue, but cls was freed by send_core_* */
     queue_destroy(queue, GNUNET_NO);
 
     /* If more data in queue, send next */
-    if (NULL != queue_head)
+    if (NULL != peer->queue_head)
     {
         struct GNUNET_PeerIdentity id;
 
+        GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "*********   more data!\n");
         GNUNET_PEER_resolve (peer->id, &id);
-        core_transmit =
+        peer->core_transmit =
             GNUNET_CORE_notify_transmit_ready(core_handle,
                                               0,
                                               0,
                                               GNUNET_TIME_UNIT_FOREVER_REL,
                                               &id,
-                                              queue_head->size,
+                                              peer->queue_head->size,
                                               &queue_send,
                                               peer);
     }
+    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "*********   return %d\n", data_size);
     return data_size;
 }
 
@@ -2797,24 +2884,36 @@ queue_send (void *cls, size_t size, void *buf)
  * @param type Type of the message.
  * @param size Size of the message.
  * @param dst Neighbor to send message to.
+ * @param t Tunnel this message belongs to.
  */
 static void
-queue_add (void *cls, uint16_t type, size_t size, struct MeshPeerInfo *dst)
+queue_add (void *cls, uint16_t type, size_t size,
+           struct MeshPeerInfo *dst, struct MeshTunnel *t)
 {
     struct MeshPeerQueue *queue;
 
+    if (t->queue_n >= t->queue_max)
+    {
+      if (NULL == t->owner)
+        GNUNET_break_op(0);       // TODO: kill connection?
+      else
+        GNUNET_break(0);
+      return;                       // Drop message
+    }
+    t->queue_n++;
     queue = GNUNET_malloc (sizeof (struct MeshPeerQueue));
     queue->cls = cls;
     queue->type = type;
     queue->size = size;
     queue->peer = dst;
-    GNUNET_CONTAINER_DLL_insert_tail (queue_head, queue_tail, queue);
-    if (NULL == core_transmit)
+    queue->tunnel = t;
+    GNUNET_CONTAINER_DLL_insert_tail (dst->queue_head, dst->queue_tail, queue);
+    if (NULL == dst->core_transmit)
     {
         struct GNUNET_PeerIdentity id;
 
         GNUNET_PEER_resolve (dst->id, &id);
-        core_transmit =
+        dst->core_transmit =
             GNUNET_CORE_notify_transmit_ready(core_handle,
                                               0,
                                               0,
@@ -2856,7 +2955,7 @@ handle_mesh_path_create (void *cls, const struct GNUNET_PeerIdentity *peer,
   MESH_TunnelNumber tid;
   struct GNUNET_MESH_ManipulatePath *msg;
   struct GNUNET_PeerIdentity *pi;
-  GNUNET_HashCode hash;
+  struct GNUNET_HashCode hash;
   struct MeshPeerPath *path;
   struct MeshPeerInfo *dest_peer_info;
   struct MeshPeerInfo *orig_peer_info;
@@ -2892,27 +2991,16 @@ handle_mesh_path_create (void *cls, const struct GNUNET_PeerIdentity *peer,
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
               "    path is for tunnel %s [%X].\n", GNUNET_i2s (pi), tid);
   t = tunnel_get (pi, tid);
-  if (NULL == t)
+  if (NULL == t) // FIXME only for INCOMING tunnels?
   {
     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "  Creating tunnel\n");
-    t = GNUNET_malloc (sizeof (struct MeshTunnel));
-    t->id.oid = GNUNET_PEER_intern (pi);
-    t->id.tid = tid;
+    t = tunnel_new (GNUNET_PEER_intern (pi), tid, NULL, 0);
+
     while (NULL != tunnel_get_incoming (next_local_tid))
       next_local_tid = (next_local_tid + 1) | GNUNET_MESH_LOCAL_TUNNEL_ID_SERV;
     t->local_tid_dest = next_local_tid++;
     next_local_tid = next_local_tid | GNUNET_MESH_LOCAL_TUNNEL_ID_SERV;
-    t->tree = tree_new (t->id.oid);
 
-    GNUNET_CRYPTO_hash (&t->id, sizeof (struct MESH_TunnelID), &hash);
-    if (GNUNET_OK !=
-        GNUNET_CONTAINER_multihashmap_put (tunnels, &hash, t,
-                                           GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_FAST))
-    {
-      tunnel_destroy (t);
-      GNUNET_break (0);
-      return GNUNET_OK;
-    }
     tunnel_reset_timeout (t);
     GNUNET_CRYPTO_hash (&t->local_tid_dest, sizeof (MESH_TunnelNumber), &hash);
     if (GNUNET_OK !=
@@ -2994,7 +3082,8 @@ handle_mesh_path_create (void *cls, const struct GNUNET_PeerIdentity *peer,
     queue_add(info,
               GNUNET_MESSAGE_TYPE_MESH_PATH_ACK,
               sizeof (struct GNUNET_MESH_PathACK),
-              info->peer);
+              info->peer,
+              t);
   }
   else
   {
@@ -3086,7 +3175,7 @@ handle_mesh_path_destroy (void *cls, const struct GNUNET_PeerIdentity *peer,
   }
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "  Own position: %u\n", own_pos);
   if (own_pos < path->length - 1)
-    send_message (message, &pi[own_pos + 1]);
+    send_message (message, &pi[own_pos + 1], t);
   else
     send_client_tunnel_disconnect(t, NULL);
 
@@ -3243,7 +3332,7 @@ handle_mesh_data_unicast (void *cls, const struct GNUNET_PeerIdentity *peer,
   }
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
               "  not for us, retransmitting...\n");
-  send_message (message, tree_get_first_hop (t->tree, pid));
+  send_message (message, tree_get_first_hop (t->tree, pid), t);
   return GNUNET_OK;
 }
 
@@ -3400,7 +3489,7 @@ handle_mesh_data_to_orig (void *cls, const struct GNUNET_PeerIdentity *peer,
     return GNUNET_OK;
   }
   GNUNET_PEER_resolve (tree_get_predecessor (t->tree), &id);
-  send_message (message, &id);
+  send_message (message, &id, t);
 
   return GNUNET_OK;
 }
@@ -3486,7 +3575,7 @@ handle_mesh_path_ack (void *cls, const struct GNUNET_PeerIdentity *peer,
     GNUNET_break (0);
     return GNUNET_OK;
   }
-  send_message (message, &id);
+  send_message (message, &id, t);
   return GNUNET_OK;
 }
 
@@ -3524,7 +3613,7 @@ static struct GNUNET_CORE_MessageHandler core_handlers[] = {
  * @return GNUNET_OK on success
  */
 static int
-deregister_app (void *cls, const GNUNET_HashCode * key, void *value)
+deregister_app (void *cls, const struct GNUNET_HashCode * key, void *value)
 {
   GNUNET_break (GNUNET_YES ==
                 GNUNET_CONTAINER_multihashmap_remove (applications, key,
@@ -3639,7 +3728,7 @@ path_refresh (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
  */
 static void
 dht_get_id_handler (void *cls, struct GNUNET_TIME_Absolute exp,
-                    const GNUNET_HashCode * key,
+                    const struct GNUNET_HashCode * key,
                     const struct GNUNET_PeerIdentity *get_path,
                     unsigned int get_path_length,
                     const struct GNUNET_PeerIdentity *put_path,
@@ -3687,20 +3776,26 @@ dht_get_id_handler (void *cls, struct GNUNET_TIME_Absolute exp,
  */
 static void
 dht_get_type_handler (void *cls, struct GNUNET_TIME_Absolute exp,
-                      const GNUNET_HashCode * key,
+                      const struct GNUNET_HashCode * key,
                       const struct GNUNET_PeerIdentity *get_path,
                       unsigned int get_path_length,
                       const struct GNUNET_PeerIdentity *put_path,
                       unsigned int put_path_length, enum GNUNET_BLOCK_Type type,
                       size_t size, const void *data)
 {
-  const struct GNUNET_PeerIdentity *pi = data;
+  const struct PBlock *pb = data;
+  const struct GNUNET_PeerIdentity *pi = &pb->id;
   struct MeshTunnel *t = cls;
   struct MeshPeerInfo *peer_info;
   struct MeshPeerPath *p;
 
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "got type DHT result!\n");
-  if (size != sizeof (struct GNUNET_PeerIdentity))
+  if (size != sizeof (struct PBlock))
+  {
+    GNUNET_break_op (0);
+    return;
+  }
+  if (ntohl(pb->type) != t->type)
   {
     GNUNET_break_op (0);
     return;
@@ -3838,7 +3933,7 @@ handle_local_new_client (void *cls, struct GNUNET_SERVER_Client *client,
   if (napps > 0)
   {
     GNUNET_MESH_ApplicationType at;
-    GNUNET_HashCode hc;
+    struct GNUNET_HashCode hc;
 
     c->apps = GNUNET_CONTAINER_multihashmap_create (napps);
     for (i = 0; i < napps; i++)
@@ -3847,7 +3942,7 @@ handle_local_new_client (void *cls, struct GNUNET_SERVER_Client *client,
       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "  app type: %u\n", at);
       GNUNET_CRYPTO_hash (&at, sizeof (at), &hc);
       /* store in clients hashmap */
-      GNUNET_CONTAINER_multihashmap_put (c->apps, &hc, c,
+      GNUNET_CONTAINER_multihashmap_put (c->apps, &hc, (void *) (long) at,
                                          GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
       /* store in global hashmap, for announcements */
       GNUNET_CONTAINER_multihashmap_put (applications, &hc, c,
@@ -3861,7 +3956,7 @@ handle_local_new_client (void *cls, struct GNUNET_SERVER_Client *client,
   if (ntypes > 0)
   {
     uint16_t u16;
-    GNUNET_HashCode hc;
+    struct GNUNET_HashCode hc;
 
     t = (uint16_t *) & a[napps];
     c->types = GNUNET_CONTAINER_multihashmap_create (ntypes);
@@ -3907,7 +4002,6 @@ handle_local_tunnel_create (void *cls, struct GNUNET_SERVER_Client *client,
   struct GNUNET_MESH_TunnelMessage *t_msg;
   struct MeshTunnel *t;
   struct MeshClient *c;
-  GNUNET_HashCode hash;
 
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "new tunnel requested\n");
 
@@ -3944,39 +4038,14 @@ handle_local_tunnel_create (void *cls, struct GNUNET_SERVER_Client *client,
     return;
   }
 
-  t = GNUNET_malloc (sizeof (struct MeshTunnel));
   while (NULL != tunnel_get_by_pi (myid, next_tid))
     next_tid = (next_tid + 1) & ~GNUNET_MESH_LOCAL_TUNNEL_ID_CLI;
-  t->id.tid = next_tid++;
+  t = tunnel_new (myid, next_tid++, c, ntohl (t_msg->tunnel_id));
   next_tid = next_tid & ~GNUNET_MESH_LOCAL_TUNNEL_ID_CLI;
-  t->id.oid = myid;
-  t->local_tid = ntohl (t_msg->tunnel_id);
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "CREATED TUNNEL %s [%x] (%x)\n",
               GNUNET_i2s (&my_full_id), t->id.tid, t->local_tid);
-  t->owner = c;
   t->peers = GNUNET_CONTAINER_multihashmap_create (32);
 
-  GNUNET_CRYPTO_hash (&t->local_tid, sizeof (MESH_TunnelNumber), &hash);
-  if (GNUNET_OK !=
-      GNUNET_CONTAINER_multihashmap_put (c->own_tunnels, &hash, t,
-                                         GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY))
-  {
-    GNUNET_break (0);
-    GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
-    return;
-  }
-
-  GNUNET_CRYPTO_hash (&t->id, sizeof (struct MESH_TunnelID), &hash);
-  if (GNUNET_OK !=
-      GNUNET_CONTAINER_multihashmap_put (tunnels, &hash, t,
-                                         GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY))
-  {
-    GNUNET_break (0);
-    GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
-    return;
-  }
-  t->tree = tree_new (myid);
-
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "new tunnel created\n");
   GNUNET_SERVER_receive_done (client, GNUNET_OK);
   return;
@@ -4199,6 +4268,126 @@ handle_local_connect_del (void *cls, struct GNUNET_SERVER_Client *client,
   return;
 }
 
+/**
+ * Handler for blacklist requests of peers in a tunnel
+ *
+ * @param cls closure
+ * @param client identification of the client
+ * @param message the actual message (PeerControl)
+ */
+static void
+handle_local_blacklist (void *cls, struct GNUNET_SERVER_Client *client,
+                          const struct GNUNET_MessageHeader *message)
+{
+  struct GNUNET_MESH_PeerControl *peer_msg;
+  struct MeshClient *c;
+  struct MeshTunnel *t;
+  MESH_TunnelNumber tid;
+
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Got a PEER BLACKLIST request\n");
+  /* Sanity check for client registration */
+  if (NULL == (c = client_get (client)))
+  {
+    GNUNET_break (0);
+    GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
+    return;
+  }
+  peer_msg = (struct GNUNET_MESH_PeerControl *) message;
+
+  /* Sanity check for message size */
+  if (sizeof (struct GNUNET_MESH_PeerControl) != ntohs (peer_msg->header.size))
+  {
+    GNUNET_break (0);
+    GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
+    return;
+  }
+
+  /* Tunnel exists? */
+  tid = ntohl (peer_msg->tunnel_id);
+  t = tunnel_get_by_local_id (c, tid);
+  if (NULL == t)
+  {
+    GNUNET_break (0);
+    GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
+    return;
+  }
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "  on tunnel %X\n", t->id.tid);
+
+  GNUNET_array_append(t->blacklisted, t->nblacklisted,
+                      GNUNET_PEER_intern(&peer_msg->peer));
+}
+
+
+/**
+ * Handler for unblacklist requests of peers in a tunnel
+ *
+ * @param cls closure
+ * @param client identification of the client
+ * @param message the actual message (PeerControl)
+ */
+static void
+handle_local_unblacklist (void *cls, struct GNUNET_SERVER_Client *client,
+                          const struct GNUNET_MessageHeader *message)
+{
+  struct GNUNET_MESH_PeerControl *peer_msg;
+  struct MeshClient *c;
+  struct MeshTunnel *t;
+  MESH_TunnelNumber tid;
+  GNUNET_PEER_Id pid;
+  unsigned int i;
+
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Got a PEER UNBLACKLIST request\n");
+  /* Sanity check for client registration */
+  if (NULL == (c = client_get (client)))
+  {
+    GNUNET_break (0);
+    GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
+    return;
+  }
+  peer_msg = (struct GNUNET_MESH_PeerControl *) message;
+
+  /* Sanity check for message size */
+  if (sizeof (struct GNUNET_MESH_PeerControl) != ntohs (peer_msg->header.size))
+  {
+    GNUNET_break (0);
+    GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
+    return;
+  }
+
+  /* Tunnel exists? */
+  tid = ntohl (peer_msg->tunnel_id);
+  t = tunnel_get_by_local_id (c, tid);
+  if (NULL == t)
+  {
+    GNUNET_break (0);
+    GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
+    return;
+  }
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "  on tunnel %X\n", t->id.tid);
+
+  /* if peer is not known, complain */
+  pid = GNUNET_PEER_search (&peer_msg->peer);
+  if (0 == pid)
+  {
+    GNUNET_break (0);
+    return;
+  }
+
+  /* search and remove from list */
+  for (i = 0; i < t->nblacklisted; i++)
+  {
+    if (t->blacklisted[i] == pid)
+    {
+      t->blacklisted[i] = t->blacklisted[t->nblacklisted - 1];
+      GNUNET_array_grow (t->blacklisted, t->nblacklisted, t->nblacklisted - 1);
+      return;
+    }
+  }
+
+  /* if peer hasn't been blacklisted, complain */
+  GNUNET_break (0);
+}
+
 
 /**
  * Handler for connection requests to new peers by type
@@ -4214,7 +4403,7 @@ handle_local_connect_by_type (void *cls, struct GNUNET_SERVER_Client *client,
   struct GNUNET_MESH_ConnectPeerByType *connect_msg;
   struct MeshClient *c;
   struct MeshTunnel *t;
-  GNUNET_HashCode hash;
+  struct GNUNET_HashCode hash;
   MESH_TunnelNumber tid;
 
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "got connect by type request\n");
@@ -4293,9 +4482,12 @@ handle_local_connect_by_type (void *cls, struct GNUNET_SERVER_Client *client,
               GNUNET_h2s (&hash));
   t->dht_get_type =
       GNUNET_DHT_get_start (dht_handle, 
-                            GNUNET_BLOCK_TYPE_TEST, &hash, 10,
+                            GNUNET_BLOCK_TYPE_MESH_PEER_BY_TYPE,
+                            &hash,
+                            10,
                             GNUNET_DHT_RO_RECORD_ROUTE |
-                            GNUNET_DHT_RO_DEMULTIPLEX_EVERYWHERE, NULL, 0,
+                            GNUNET_DHT_RO_DEMULTIPLEX_EVERYWHERE,
+                            NULL, 0,
                             &dht_get_type_handler, t);
 
   GNUNET_SERVER_receive_done (client, GNUNET_OK);
@@ -4567,6 +4759,12 @@ static struct GNUNET_SERVER_MessageHandler client_handlers[] = {
   {&handle_local_connect_del, NULL,
    GNUNET_MESSAGE_TYPE_MESH_LOCAL_PEER_DEL,
    sizeof (struct GNUNET_MESH_PeerControl)},
+  {&handle_local_blacklist, NULL,
+   GNUNET_MESSAGE_TYPE_MESH_LOCAL_PEER_BLACKLIST,
+   sizeof (struct GNUNET_MESH_PeerControl)},
+  {&handle_local_unblacklist, NULL,
+   GNUNET_MESSAGE_TYPE_MESH_LOCAL_PEER_UNBLACKLIST,
+   sizeof (struct GNUNET_MESH_PeerControl)},
   {&handle_local_connect_by_type, NULL,
    GNUNET_MESSAGE_TYPE_MESH_LOCAL_PEER_ADD_BY_TYPE,
    sizeof (struct GNUNET_MESH_ConnectPeerByType)},
@@ -4659,14 +4857,14 @@ core_disconnect (void *cls, const struct GNUNET_PeerIdentity *peer)
     GNUNET_break (0);
     return;
   }
-  q = queue_head;
+  q = pi->queue_head;
   while (NULL != q)
   {
       n = q->next;
       if (q->peer == pi)
       {
         /* try to reroute this traffic instead */
-        queue_destroy(queue_head, GNUNET_YES);
+        queue_destroy(q, GNUNET_YES);
       }
       q = n;
   }
@@ -4693,7 +4891,7 @@ core_disconnect (void *cls, const struct GNUNET_PeerIdentity *peer)
  *         GNUNET_NO if not.
  */
 static int
-shutdown_tunnel (void *cls, const GNUNET_HashCode * key, void *value)
+shutdown_tunnel (void *cls, const struct GNUNET_HashCode * key, void *value)
 {
   struct MeshTunnel *t = value;
 
@@ -4711,19 +4909,19 @@ shutdown_tunnel (void *cls, const GNUNET_HashCode * key, void *value)
  *         GNUNET_NO if not.
  */
 static int
-shutdown_peer (void *cls, const GNUNET_HashCode * key, void *value)
+shutdown_peer (void *cls, const struct GNUNET_HashCode * key, void *value)
 {
   struct MeshPeerInfo *p = value;
   struct MeshPeerQueue *q;
   struct MeshPeerQueue *n;
 
-  q = queue_head;
+  q = p->queue_head;
   while (NULL != q)
   {
       n = q->next;
       if (q->peer == p)
       {
-        queue_destroy(queue_head, GNUNET_YES);
+        queue_destroy(q, GNUNET_YES);
       }
       q = n;
   }
@@ -4785,7 +4983,6 @@ run (void *cls, struct GNUNET_SERVER_Handle *server,
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "starting to run\n");
   server_handle = server;
   core_handle = GNUNET_CORE_connect (c, /* Main configuration */
-                                     1,   /* queue size */
                                      NULL,      /* Closure passed to MESH functions */
                                      &core_init,        /* Call core_init once connected */
                                      &core_connect,     /* Handle connects */
@@ -4827,12 +5024,12 @@ run (void *cls, struct GNUNET_SERVER_Handle *server,
                       &my_full_id.hashPubKey);
   myid = GNUNET_PEER_intern (&my_full_id);
 
-// //   transport_handle = GNUNET_TRANSPORT_connect(c,
-// //                                               &my_full_id,
-// //                                               NULL,
-// //                                               NULL,
-// //                                               NULL,
-// //                                               NULL);
+//   transport_handle = GNUNET_TRANSPORT_connect(c,
+//                                               &my_full_id,
+//                                               NULL,
+//                                               NULL,
+//                                               NULL,
+//                                               NULL);
 
   dht_handle = GNUNET_DHT_connect (c, 64);
   if (dht_handle == NULL)