- use blocks for checking by type results
[oweals/gnunet.git] / src / mesh / gnunet-service-mesh_new.c
index 34a48dabed3609e4ecec4031e6dc673fd7f07343..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__)
@@ -137,9 +140,9 @@ struct MeshPeerQueue
   struct MeshPeerInfo *peer;
 
     /**
-     * Handle to stop queued transmission
+     * Tunnel this message belongs to.
      */
-  struct GNUNET_CORE_TransmitHandle *core_transmit;
+  struct MeshTunnel *tunnel;
 
     /**
      * Pointer to info stucture used as cls.
@@ -150,6 +153,11 @@ struct MeshPeerQueue
      * Type of message
      */
   uint16_t type;
+
+    /**
+     * Size of the message
+     */
+  size_t size;
 };
 
 
@@ -168,9 +176,6 @@ struct MeshTransmissionDescriptor
     /** Ultimate destination of the packet */
   GNUNET_PEER_Id destination;
 
-    /** Which handler was used to request the transmission */
-  struct MeshPeerQueue *queue;
-
     /** Data descriptor */
   struct MeshData* mesh_data;
 };
@@ -215,16 +220,6 @@ struct MeshPeerInfo
      * Closure given to the DHT GET
      */
   struct MeshPathInfo *dhtgetcls;
-  
-    /**
-     * Transmission queue to this peer, head
-     */
-  struct MeshPeerQueue *queue_head;
-
-    /**
-     * Transmission queue to this peer, tail
-     */
-  struct MeshPeerQueue *queue_tail;
 
     /**
      * Array of tunnels this peer participates in
@@ -238,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
@@ -326,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
      */
@@ -377,10 +375,19 @@ struct MeshTunnel
   unsigned int nignore;
 
     /**
-     * Messages ready to transmit
+     * Blacklisted peers
      */
-  struct MeshQueue *queue_head;
-  struct MeshQueue *queue_tail;
+  GNUNET_PEER_Id *blacklisted;
+
+    /**
+     * Number of elements in blacklisted
+     */
+  unsigned int nblacklisted;
+
+  /**
+   * Bloomfilter (for peer identities) to stop circular routes
+   */
+  char bloomfilter[MESH_BLOOM_SIZE];
 
   /**
    * Tunnel paths
@@ -431,11 +438,6 @@ struct MeshPathInfo
    * Path itself
    */
   struct MeshPeerPath *path;
-
-  /**
-   * Position in peer's transmit queue
-   */
-  struct MeshPeerQueue *queue;
 };
 
 
@@ -551,15 +553,15 @@ static struct GNUNET_CONTAINER_MultiHashMap *incoming_tunnels;
  */
 static struct GNUNET_CONTAINER_MultiHashMap *peers;
 
-/**
- * Handle to communicate with core
+/*
+ * Handle to communicate with transport
  */
-static struct GNUNET_CORE_Handle *core_handle;
+// static struct GNUNET_TRANSPORT_Handle *transport_handle;
 
 /**
- * Handle to communicate with transport
+ * Handle to communicate with core
  */
-// static struct GNUNET_TRANSPORT_Handle *transport_handle;
+static struct GNUNET_CORE_Handle *core_handle;
 
 /**
  * Handle to use DHT
@@ -655,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);
@@ -708,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;
@@ -719,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 */
@@ -749,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,
@@ -834,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);
@@ -923,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 ==
@@ -947,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)
   {
@@ -1047,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);
@@ -1160,37 +1179,6 @@ send_client_tunnel_disconnect (struct MeshTunnel *t, struct MeshClient *c)
 }
 
 
-/**
- * Function called to notify a client about the socket
- * being ready to queue more data.  "buf" will be
- * NULL and "size" zero if the socket was closed for
- * writing in the meantime.
- *
- * @param cls closure
- * @param size number of bytes available in buf
- * @param buf where the callee should write the message
- * @return number of bytes written to buf
- */
-static size_t
-send_core_create_path (void *cls, size_t size, void *buf);
-
-
-/**
- * Function called to notify a client about the socket
- * being ready to queue more data.  "buf" will be
- * NULL and "size" zero if the socket was closed for
- * writing in the meantime.
- *
- * @param cls closure (data itself)
- * @param size number of bytes available in buf
- * @param buf where the callee should write the message
- *
- * @return number of bytes written to buf
- */
-static size_t
-send_core_data_multicast (void *cls, size_t size, void *buf);
-
-
 /**
  * Decrements the reference counter and frees all resources if needed
  *
@@ -1223,61 +1211,6 @@ data_descriptor_decrement_multicast (struct MeshData *mesh_data)
 }
 
 
-/**
- * Cancel a core transmission that was already queued, free all resources
- * associated to the request and cancel all external requests (core, ...).
- *
- * @param queue Queue handler to cancel.
- */
-static void
-peer_info_cancel_transmission (struct MeshPeerQueue *queue)
-{
-  struct MeshTransmissionDescriptor *dd;
-  struct MeshPathInfo *path_info;
-  struct MeshPeerInfo *peer;
-
-  peer = queue->peer;
-  if (NULL != queue->core_transmit)
-  {
-#if MESH_DEBUG
-    {
-      struct GNUNET_PeerIdentity id;
-
-      GNUNET_PEER_resolve (peer->id, &id);
-      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-                  "  Cancelling data transmission at %s\n",
-                  GNUNET_i2s (&id));
-      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "   message type %u\n",
-                  queue->type);
-    }
-#endif
-    GNUNET_CORE_notify_transmit_ready_cancel (queue->core_transmit);
-  }
-  /* TODO: notify that transmission has failed */
-  switch (queue->type)
-  {
-  case GNUNET_MESSAGE_TYPE_MESH_MULTICAST:
-  case GNUNET_MESSAGE_TYPE_MESH_UNICAST:
-  case GNUNET_MESSAGE_TYPE_MESH_TO_ORIGIN:
-    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "   type payload\n");
-    dd = queue->cls;
-    data_descriptor_decrement_multicast (dd->mesh_data);
-    break;
-  case GNUNET_MESSAGE_TYPE_MESH_PATH_CREATE:
-    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "   type create path\n");
-    path_info = queue->cls;
-    path_destroy (path_info->path);
-    break;
-  default:
-    GNUNET_break (0);
-    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "   type unknown!\n");
-  }
-  GNUNET_free_non_null (queue->cls);
-  GNUNET_CONTAINER_DLL_remove (peer->queue_head, peer->queue_tail, queue);
-  GNUNET_free(queue);
-}
-
-
 /**
  * Retrieve the MeshPeerInfo stucture associated with the peer, create one
  * and insert it in the appropiate structures if the peer is not known yet.
@@ -1334,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;
@@ -1355,6 +1288,19 @@ peer_info_delete_tunnel (void *cls, const GNUNET_HashCode * key, void *value)
 }
 
 
+/**
+ * Queue and pass message to core when possible.
+ *
+ * @param cls Closure (type dependant).
+ * @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, struct MeshTunnel *t);
+
 /**
   * Core callback to write a pre-constructed data packet to core buffer
   *
@@ -1369,52 +1315,39 @@ send_core_data_raw (void *cls, size_t size, void *buf)
 {
   struct MeshTransmissionDescriptor *info = cls;
   struct GNUNET_MessageHeader *msg;
-  struct MeshPeerQueue *queue;
   size_t total_size;
 
   GNUNET_assert (NULL != info);
   GNUNET_assert (NULL != info->mesh_data);
   msg = (struct GNUNET_MessageHeader *) info->mesh_data->data;
   total_size = ntohs (msg->size);
-  queue = info->queue;
 
   if (total_size > size)
   {
-    struct GNUNET_PeerIdentity id;
-
-    GNUNET_PEER_resolve (info->peer->id, &id);
-    queue->core_transmit =
-        GNUNET_CORE_notify_transmit_ready (core_handle, 0, 100,
-                                           GNUNET_TIME_UNIT_FOREVER_REL, &id,
-                                           size, &send_core_data_raw, info);
+    GNUNET_break (0);
     return 0;
   }
   memcpy (buf, msg, total_size);
   GNUNET_free (info->mesh_data);
   GNUNET_free (info);
-  GNUNET_CONTAINER_DLL_remove (queue->peer->queue_head,
-                               queue->peer->queue_tail,
-                               queue);
-  GNUNET_free (queue);
   return total_size;
 }
 
 
 /**
- * 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 MeshPeerQueue *queue;
   struct MeshPeerInfo *neighbor;
   struct MeshPeerPath *p;
   size_t size;
@@ -1443,19 +1376,12 @@ send_message (const struct GNUNET_MessageHeader *message,
     GNUNET_free (info);
     return;
   }
-  queue = GNUNET_malloc (sizeof(struct MeshPeerQueue));
-  queue->peer = neighbor;
-  info->queue = queue;
   info->peer = neighbor;
-  queue->type = GNUNET_MESSAGE_TYPE_MESH_UNICAST;
-  queue->cls = info;
-  queue->core_transmit =
-      GNUNET_CORE_notify_transmit_ready (core_handle, 0, 100,
-                                         GNUNET_TIME_UNIT_FOREVER_REL, peer,
-                                         size, &send_core_data_raw, info);
-  GNUNET_CONTAINER_DLL_insert (neighbor->queue_head,
-                               neighbor->queue_tail,
-                               queue);
+  queue_add (info,
+             GNUNET_MESSAGE_TYPE_MESH_UNICAST,
+             size,
+             neighbor,
+             t);
 }
 
 
@@ -1474,7 +1400,6 @@ send_create_path (struct MeshPeerInfo *peer, struct MeshPeerPath *p,
   struct GNUNET_PeerIdentity id;
   struct MeshPathInfo *path_info;
   struct MeshPeerInfo *neighbor;
-  struct MeshPeerQueue *queue;
 
   unsigned int i;
 
@@ -1500,28 +1425,17 @@ send_create_path (struct MeshPeerInfo *peer, struct MeshPeerPath *p,
   }
   GNUNET_PEER_resolve (p->peers[i + 1], &id);
 
-  queue = GNUNET_malloc (sizeof(struct MeshPeerQueue));
   path_info = GNUNET_malloc (sizeof (struct MeshPathInfo));
   path_info->path = p;
   path_info->t = t;
   neighbor = peer_info_get (&id);
   path_info->peer = neighbor;
-  path_info->queue = queue;
-  queue->type = GNUNET_MESSAGE_TYPE_MESH_PATH_CREATE;
-  queue->cls = path_info;
-  queue->core_transmit =
-    GNUNET_CORE_notify_transmit_ready (core_handle,     /* handle */
-                                       0,       /* cork */
-                                       0,       /* priority */
-                                       GNUNET_TIME_UNIT_FOREVER_REL,    /* timeout */
-                                       &id,     /* target */
-                                       sizeof (struct GNUNET_MESH_ManipulatePath) +
-                                         (p->length * sizeof (struct GNUNET_PeerIdentity)),  /*size */
-                                       &send_core_create_path,  /* callback */
-                                       path_info);      /* cls */
-  GNUNET_CONTAINER_DLL_insert (neighbor->queue_head,
-                               neighbor->queue_tail,
-                               queue);
+  queue_add (path_info,
+             GNUNET_MESSAGE_TYPE_MESH_PATH_CREATE,
+             sizeof (struct GNUNET_MESH_ManipulatePath) +
+                (p->length * sizeof (struct GNUNET_PeerIdentity)),
+             neighbor,
+             t);
 }
 
 
@@ -1560,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);
 }
@@ -1597,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);
@@ -1625,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. */
@@ -1672,7 +1588,6 @@ peer_info_destroy (struct MeshPeerInfo *pi)
   struct GNUNET_PeerIdentity id;
   struct MeshPeerPath *p;
   struct MeshPeerPath *nextp;
-  unsigned int i;
 
   GNUNET_PEER_resolve (pi->id, &id);
   GNUNET_PEER_change_rc (pi->id, -1);
@@ -1689,10 +1604,6 @@ peer_info_destroy (struct MeshPeerInfo *pi)
     GNUNET_DHT_get_stop (pi->dhtget);
     GNUNET_free (pi->dhtgetcls);
   }
-  for (i = 0; i < CORE_QUEUE_SIZE; i++)
-  {
-    // FIXME CORE_QUEUE_SIZE
-  }
   p = pi->path_head;
   while (NULL != p)
   {
@@ -2039,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);
@@ -2064,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);
@@ -2084,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;
@@ -2325,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;
@@ -2344,7 +2255,6 @@ tunnel_send_multicast_iterator (void *cls, GNUNET_PEER_Id neighbor_id)
   struct MeshData *mdata = cls;
   struct MeshTransmissionDescriptor *info;
   struct GNUNET_PeerIdentity neighbor;
-  struct MeshPeerQueue *queue;
 
   info = GNUNET_malloc (sizeof (struct MeshTransmissionDescriptor));
 
@@ -2356,19 +2266,11 @@ tunnel_send_multicast_iterator (void *cls, GNUNET_PEER_Id neighbor_id)
               GNUNET_i2s (&neighbor));
   info->peer = peer_info_get (&neighbor);
   GNUNET_assert (NULL != info->peer);
-  queue = GNUNET_malloc (sizeof(struct MeshPeerQueue));
-  info->queue = queue;
-  queue->cls = info;
-  queue->type = GNUNET_MESSAGE_TYPE_MESH_MULTICAST;
-  queue->core_transmit =
-      GNUNET_CORE_notify_transmit_ready (core_handle, 0, 0,
-                                         GNUNET_TIME_UNIT_FOREVER_REL,
-                                         &neighbor, info->mesh_data->data_len,
-                                         &send_core_data_multicast, info);
-  queue->peer = info->peer;
-  GNUNET_CONTAINER_DLL_insert (queue->peer->queue_head,
-                               queue->peer->queue_tail,
-                               queue);
+  queue_add(info,
+            GNUNET_MESSAGE_TYPE_MESH_MULTICAST,
+            info->mesh_data->data_len,
+            info->peer,
+            mdata->t);
 }
 
 /**
@@ -2469,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;
 
@@ -2539,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);
@@ -2561,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.
@@ -2590,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;
@@ -2649,10 +2596,7 @@ tunnel_reset_timeout (struct MeshTunnel *t)
 /******************************************************************************/
 
 /**
- * Function called to notify a client about the socket
- * being ready to queue more data.  "buf" will be
- * NULL and "size" zero if the socket was closed for
- * writing in the meantime.
+ * Function to send a create path packet to a peer.
  *
  * @param cls closure
  * @param size number of bytes available in buf
@@ -2660,12 +2604,11 @@ tunnel_reset_timeout (struct MeshTunnel *t)
  * @return number of bytes written to buf
  */
 static size_t
-send_core_create_path (void *cls, size_t size, void *buf)
+send_core_path_create (void *cls, size_t size, void *buf)
 {
   struct MeshPathInfo *info = cls;
   struct GNUNET_MESH_ManipulatePath *msg;
   struct GNUNET_PeerIdentity *peer_ptr;
-  struct MeshPeerInfo *peer = info->peer;
   struct MeshTunnel *t = info->t;
   struct MeshPeerPath *p = info->path;
   size_t size_needed;
@@ -2678,17 +2621,7 @@ send_core_create_path (void *cls, size_t size, void *buf)
 
   if (size < size_needed || NULL == buf)
   {
-    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "create path retransmit!\n");
-    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "  buf:  %p\n", buf);
-    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "  size: (%u/%u)\n", size,
-                size_needed);
-    info->queue->core_transmit =
-        GNUNET_CORE_notify_transmit_ready (core_handle, 0, 0,
-                                           GNUNET_TIME_UNIT_FOREVER_REL,
-                                           tree_get_first_hop (t->tree,
-                                                               peer->id),
-                                           size_needed, &send_core_create_path,
-                                           info);
+    GNUNET_break (0);
     return 0;
   }
   msg = (struct GNUNET_MESH_ManipulatePath *) buf;
@@ -2703,10 +2636,6 @@ send_core_create_path (void *cls, size_t size, void *buf)
   }
 
   path_destroy (p);
-  GNUNET_CONTAINER_DLL_remove(info->queue->peer->queue_head,
-                              info->queue->peer->queue_tail,
-                              info->queue);
-  GNUNET_free (info->queue);
   GNUNET_free (info);
 
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
@@ -2716,10 +2645,7 @@ send_core_create_path (void *cls, size_t size, void *buf)
 
 
 /**
- * Function called to notify a client about the socket
- * being ready to queue more data.  "buf" will be
- * NULL and "size" zero if the socket was closed for
- * writing in the meantime.
+ * Fill the core buffer 
  *
  * @param cls closure (data itself)
  * @param size number of bytes available in buf
@@ -2741,18 +2667,7 @@ send_core_data_multicast (void *cls, size_t size, void *buf)
 
   if (total_size > size)
   {
-    /* Retry */
-    struct GNUNET_PeerIdentity id;
-
-    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-                "Multicast: retransmitting... (%u/%u)\n", size,
-                total_size);
-    GNUNET_PEER_resolve (info->peer->id, &id);
-    info->queue->core_transmit =
-        GNUNET_CORE_notify_transmit_ready (core_handle, 0, 0,
-                                           GNUNET_TIME_UNIT_FOREVER_REL, &id,
-                                           total_size,
-                                           &send_core_data_multicast, info);
+    GNUNET_break (0);
     return 0;
   }
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, " copying data...\n");
@@ -2781,10 +2696,6 @@ send_core_data_multicast (void *cls, size_t size, void *buf)
 #endif
   data_descriptor_decrement_multicast (info->mesh_data);
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "freeing info...\n");
-  GNUNET_CONTAINER_DLL_remove(info->queue->peer->queue_head,
-                              info->queue->peer->queue_tail,
-                              info->queue);
-  GNUNET_free (info->queue);
   GNUNET_free (info);
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "return %u\n", total_size);
   return total_size;
@@ -2792,10 +2703,7 @@ send_core_data_multicast (void *cls, size_t size, void *buf)
 
 
 /**
- * Function called to notify a client about the socket
- * being ready to queue more data.  "buf" will be
- * NULL and "size" zero if the socket was closed for
- * writing in the meantime.
+ * Creates a path ack message in buf and frees all unused resources.
  *
  * @param cls closure (MeshTransmissionDescriptor)
  * @param size number of bytes available in buf
@@ -2820,13 +2728,6 @@ send_core_path_ack (void *cls, size_t size, void *buf)
   msg->tid = htonl (info->origin->tid);
   msg->peer_id = my_full_id;
 
-  if (info->queue)
-  {
-    GNUNET_CONTAINER_DLL_remove(info->queue->peer->queue_head,
-                                info->queue->peer->queue_tail,
-                                info->queue);
-    GNUNET_free (info->queue);
-  }
   GNUNET_free (info);
   /* TODO add signature */
 
@@ -2835,6 +2736,196 @@ send_core_path_ack (void *cls, size_t size, void *buf)
 }
 
 
+/**
+ * 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 MeshTransmissionDescriptor *dd;
+  struct MeshPathInfo *path_info;
+
+  if (GNUNET_YES == clear_cls)
+  {
+    switch (queue->type)
+    {
+    case GNUNET_MESSAGE_TYPE_MESH_UNICAST:
+    case GNUNET_MESSAGE_TYPE_MESH_MULTICAST:
+        GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "   type payload\n");
+        dd = queue->cls;
+        data_descriptor_decrement_multicast (dd->mesh_data);
+        break;
+    case GNUNET_MESSAGE_TYPE_MESH_PATH_CREATE:
+        GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "   type create path\n");
+        path_info = queue->cls;
+        path_destroy (path_info->path);
+        break;
+    default:
+        GNUNET_break (0);
+        GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "   type unknown!\n");
+    }
+    GNUNET_free_non_null (queue->cls);
+  }
+  GNUNET_CONTAINER_DLL_remove (queue->peer->queue_head,
+                               queue->peer->queue_tail,
+                               queue);
+  GNUNET_free (queue);
+}
+
+
+/**
+  * Core callback to write a queued packet to core buffer
+  *
+  * @param cls Closure (peer info).
+  * @param size Number of bytes available in buf.
+  * @param buf Where the to write the message.
+  *
+  * @return number of bytes written to buf
+  */
+static size_t
+queue_send (void *cls, size_t size, void *buf)
+{
+    struct MeshPeerInfo *peer = cls;
+    struct MeshPeerQueue *queue;
+    size_t data_size;
+
+    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)
+    {
+        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)
+    {
+        struct GNUNET_PeerIdentity id;
+
+        GNUNET_PEER_resolve (peer->id, &id);
+        peer->core_transmit =
+            GNUNET_CORE_notify_transmit_ready(core_handle,
+                                              0,
+                                              0,
+                                              GNUNET_TIME_UNIT_FOREVER_REL,
+                                              &id,
+                                              queue->size,
+                                              &queue_send,
+                                              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_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");
+            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 != peer->queue_head)
+    {
+        struct GNUNET_PeerIdentity id;
+
+        GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "*********   more data!\n");
+        GNUNET_PEER_resolve (peer->id, &id);
+        peer->core_transmit =
+            GNUNET_CORE_notify_transmit_ready(core_handle,
+                                              0,
+                                              0,
+                                              GNUNET_TIME_UNIT_FOREVER_REL,
+                                              &id,
+                                              peer->queue_head->size,
+                                              &queue_send,
+                                              peer);
+    }
+    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "*********   return %d\n", data_size);
+    return data_size;
+}
+
+
+/**
+ * Queue and pass message to core when possible.
+ *
+ * @param cls Closure (type dependant).
+ * @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, 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;
+    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);
+        dst->core_transmit =
+            GNUNET_CORE_notify_transmit_ready(core_handle,
+                                              0,
+                                              0,
+                                              GNUNET_TIME_UNIT_FOREVER_REL,
+                                              &id,
+                                              size,
+                                              &queue_send,
+                                              dst);
+    }
+}
+
+
 /******************************************************************************/
 /********************      MESH NETWORK HANDLERS     **************************/
 /******************************************************************************/
@@ -2864,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;
@@ -2900,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 !=
@@ -2981,7 +3061,6 @@ handle_mesh_path_create (void *cls, const struct GNUNET_PeerIdentity *peer,
   {
     /* It is for us! Send ack. */
     struct MeshTransmissionDescriptor *info;
-    struct MeshPeerQueue *queue;
 
     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "  It's for us!\n");
     peer_info_add_path_to_origin (orig_peer_info, path, GNUNET_NO);
@@ -2996,24 +3075,15 @@ handle_mesh_path_create (void *cls, const struct GNUNET_PeerIdentity *peer,
                                                      peer_info_get
                                                      (&my_full_id),
                                                      GNUNET_CONTAINER_MULTIHASHMAPOPTION_REPLACE));
-    /* FIXME use send_message */
     info = GNUNET_malloc (sizeof (struct MeshTransmissionDescriptor));
     info->origin = &t->id;
     info->peer = GNUNET_CONTAINER_multihashmap_get (peers, &peer->hashPubKey);
     GNUNET_assert (NULL != info->peer);
-    queue = GNUNET_malloc (sizeof (struct MeshPeerQueue));
-    info->queue = queue;
-    queue->peer = info->peer;
-    queue->type = GNUNET_MESSAGE_TYPE_MESH_PATH_ACK;
-    queue->cls = info;
-    queue->core_transmit =
-        GNUNET_CORE_notify_transmit_ready (core_handle, 0, 10,
-                                           GNUNET_TIME_UNIT_FOREVER_REL, peer,
-                                           sizeof (struct GNUNET_MESH_PathACK),
-                                           &send_core_path_ack, info);
-    GNUNET_CONTAINER_DLL_insert (queue->peer->queue_head,
-                                 queue->peer->queue_tail,
-                                 queue);
+    queue_add(info,
+              GNUNET_MESSAGE_TYPE_MESH_PATH_ACK,
+              sizeof (struct GNUNET_MESH_PathACK),
+              info->peer,
+              t);
   }
   else
   {
@@ -3105,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);
 
@@ -3262,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;
 }
 
@@ -3419,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;
 }
@@ -3505,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;
 }
 
@@ -3543,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,
@@ -3658,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,
@@ -3706,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;
@@ -3857,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++)
@@ -3866,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,
@@ -3880,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);
@@ -3926,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");
 
@@ -3963,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;
@@ -4218,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
@@ -4233,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");
@@ -4312,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);
@@ -4586,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)},
@@ -4668,7 +4847,8 @@ static void
 core_disconnect (void *cls, const struct GNUNET_PeerIdentity *peer)
 {
   struct MeshPeerInfo *pi;
-  unsigned int i;
+  struct MeshPeerQueue *q;
+  struct MeshPeerQueue *n;
 
   DEBUG_CONN ("Peer disconnected\n");
   pi = GNUNET_CONTAINER_multihashmap_get (peers, &peer->hashPubKey);
@@ -4677,10 +4857,16 @@ core_disconnect (void *cls, const struct GNUNET_PeerIdentity *peer)
     GNUNET_break (0);
     return;
   }
-  for (i = 0; i < CORE_QUEUE_SIZE; i++)
+  q = pi->queue_head;
+  while (NULL != q)
   {
-    /* TODO: notify that the transmission failed */
-    // FIXME CORE_QUEUE_SIZE
+      n = q->next;
+      if (q->peer == pi)
+      {
+        /* try to reroute this traffic instead */
+        queue_destroy(q, GNUNET_YES);
+      }
+      q = n;
   }
   peer_info_remove_path (pi, pi->id, myid);
   if (myid == pi->id)
@@ -4705,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;
 
@@ -4723,10 +4909,22 @@ 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 = p->queue_head;
+  while (NULL != q)
+  {
+      n = q->next;
+      if (q->peer == p)
+      {
+        queue_destroy(q, GNUNET_YES);
+      }
+      q = n;
+  }
   peer_info_destroy (p);
   return GNUNET_YES;
 }
@@ -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)