X-Git-Url: https://git.librecmc.org/?a=blobdiff_plain;f=src%2Fmesh%2Fmesh_api.c;h=6bdc2569508af09c37876b2aabb46e988428d5cf;hb=5d74fb965c6d619c323789da837e05a4b9c5def4;hp=101510d40b5806e950baf0693b1ddeb60d990b53;hpb=4649d3d4a4190e27a42d571c7d7ee7e17792dc51;p=oweals%2Fgnunet.git

diff --git a/src/mesh/mesh_api.c b/src/mesh/mesh_api.c
index 101510d40..6bdc25695 100644
--- a/src/mesh/mesh_api.c
+++ b/src/mesh/mesh_api.c
@@ -21,21 +21,15 @@
  * @author Bartlomiej Polot
  *
  * STRUCTURE:
- * - CONSTANTS
  * - DATA STRUCTURES
+ * - DECLARATIONS
  * - AUXILIARY FUNCTIONS
  * - RECEIVE HANDLERS
  * - SEND FUNCTIONS
  * - API CALL DEFINITIONS
+ *
+ * TODO: add regex to reconnect
  */
-#ifdef __cplusplus
-extern "C"
-{
-#if 0                           /* keep Emacsens' auto-indent happy */
-}
-#endif
-#endif
-
 #include "platform.h"
 #include "gnunet_common.h"
 #include "gnunet_client_lib.h"
@@ -45,10 +39,9 @@ extern "C"
 #include "mesh.h"
 #include "mesh_protocol.h"
 
-#define MESH_API_DEBUG GNUNET_YES
-
 #define LOG(kind,...) GNUNET_log_from (kind, "mesh-api",__VA_ARGS__)
 
+
 /******************************************************************************/
 /************************      DATA STRUCTURES     ****************************/
 /******************************************************************************/
@@ -70,7 +63,7 @@ struct GNUNET_MESH_TransmitHandle
   struct GNUNET_MESH_TransmitHandle *prev;
 
     /**
-     * Tunnel this message is sent over (may be NULL for control messages).
+     * Tunnel this message is sent on / for (may be NULL for control messages).
      */
   struct GNUNET_MESH_Tunnel *tunnel;
 
@@ -100,12 +93,6 @@ struct GNUNET_MESH_TransmitHandle
      */
   GNUNET_SCHEDULER_TaskIdentifier timeout_task;
 
-    /**
-     * Priority of the message.  The queue is sorted by priority,
-     * control messages have the maximum priority (UINT32_MAX).
-     */
-  uint32_t priority;
-
     /**
      * Target of the message, 0 for multicast.  This field
      * is only valid if 'notify' is non-NULL.
@@ -144,9 +131,13 @@ struct GNUNET_MESH_Handle
   const GNUNET_MESH_ApplicationType *applications;
 
     /**
-     * Double linked list of the tunnels this client is connected to.
+     * Double linked list of the tunnels this client is connected to, head.
      */
   struct GNUNET_MESH_Tunnel *tunnels_head;
+
+    /**
+     * Double linked list of the tunnels this client is connected to, tail.
+     */
   struct GNUNET_MESH_Tunnel *tunnels_tail;
 
     /**
@@ -170,18 +161,29 @@ struct GNUNET_MESH_Handle
   void *cls;
 
     /**
-     * Messages to send to the service
+     * Messages to send to the service, head.
      */
   struct GNUNET_MESH_TransmitHandle *th_head;
+
+    /**
+     * Messages to send to the service, tail.
+     */
   struct GNUNET_MESH_TransmitHandle *th_tail;
 
     /**
      * tid of the next tunnel to create (to avoid reusing IDs often)
      */
   MESH_TunnelNumber next_tid;
+
+    /**
+     * Number of handlers in the handlers array.
+     */
   unsigned int n_handlers;
+
+    /**
+     * Number of applications in the applications array.
+     */
   unsigned int n_applications;
-  unsigned int max_queue_size;
 
     /**
      * Have we started the task to receive messages from the service
@@ -189,11 +191,6 @@ struct GNUNET_MESH_Handle
      */
   int in_receive;
 
-    /**
-     * Number of packets queued
-     */
-  unsigned int npackets;
-
   /**
    * Configuration given by the client, in case of reconnection
    */
@@ -203,6 +200,11 @@ struct GNUNET_MESH_Handle
    * Time to the next reconnect in case one reconnect fails
    */
   struct GNUNET_TIME_Relative reconnect_time;
+  
+  /**
+   * Task for trying to reconnect.
+   */
+  GNUNET_SCHEDULER_TaskIdentifier reconnect_task;
 };
 
 
@@ -236,9 +238,13 @@ struct GNUNET_MESH_Tunnel
 {
 
     /**
-     * DLL
+     * DLL next
      */
   struct GNUNET_MESH_Tunnel *next;
+
+    /**
+     * DLL prev
+     */
   struct GNUNET_MESH_Tunnel *prev;
 
     /**
@@ -267,7 +273,7 @@ struct GNUNET_MESH_Tunnel
   MESH_TunnelNumber tid;
 
     /**
-     * Owner of the tunnel
+     * Owner of the tunnel. 0 if the tunnel is the local client.
      */
   GNUNET_PEER_Id owner;
 
@@ -292,22 +298,101 @@ struct GNUNET_MESH_Tunnel
   unsigned int npeers;
 
     /**
-     * Number of packets queued in this tunnel
+     * Size of packet queued in this tunnel
      */
-  unsigned int npackets;
+  unsigned int packet_size;
 
     /**
      * Number of applications requested this tunnel
      */
   unsigned int napps;
 
+    /**
+     * Is the tunnel throttled to the slowest peer?
+     */
+  int speed_min;
+
+    /**
+     * Is the tunnel allowed to buffer?
+     */
+  int buffering;
+
+    /**
+     * Next packet PID.
+     */
+  uint32_t pid;
+
+    /**
+     * Maximum allowed PID.
+     */
+  uint32_t max_pid;
+
+
 };
 
 
+/******************************************************************************/
+/***********************         DECLARATIONS         *************************/
+/******************************************************************************/
+
+/**
+ * Function called to send a message to the service.
+ * "buf" will be NULL and "size" zero if the socket was closed for writing in
+ * the meantime.
+ *
+ * @param cls closure, the mesh handle
+ * @param size number of bytes available in buf
+ * @param buf where the callee should write the connect message
+ * @return number of bytes written to buf
+ */
+static size_t
+send_callback (void *cls, size_t size, void *buf);
+
+
 /******************************************************************************/
 /***********************     AUXILIARY FUNCTIONS      *************************/
 /******************************************************************************/
 
+/**
+ * Check if transmission is a payload packet.
+ *
+ * @param th Transmission handle.
+ *
+ * @return GNUNET_YES if it is a payload packet,
+ *         GNUNET_NO if it is a mesh management packet.
+ */
+static int
+th_is_payload (struct GNUNET_MESH_TransmitHandle *th)
+{
+  return (th->notify != NULL) ? GNUNET_YES : GNUNET_NO;
+}
+
+
+/**
+ * Check whether there is any message ready in the queue and find the size.
+ * 
+ * @param h Mesh handle.
+ * 
+ * @return The size of the first ready message in the queue,
+ *         0 if there is none.
+ */
+static size_t
+message_ready_size (struct GNUNET_MESH_Handle *h)
+{
+  struct GNUNET_MESH_TransmitHandle *th;
+  struct GNUNET_MESH_Tunnel *t;
+
+  for (th = h->th_head; NULL != th; th = th->next)
+  {
+    t = th->tunnel;
+    if (GNUNET_NO == th_is_payload (th) ||
+        (t->max_pid > t->pid || PID_OVERFLOW (t->pid, t->max_pid)))
+      return th->size;
+  }
+  return 0;
+}
+
+
 /**
  * Get the tunnel handler for the tunnel specified by id from the given handle
  * @param h Mesh handle
@@ -358,6 +443,7 @@ create_tunnel (struct GNUNET_MESH_Handle *h, MESH_TunnelNumber tid)
   {
     t->tid = tid;
   }
+  t->max_pid = 1;
   return t;
 }
 
@@ -370,52 +456,33 @@ create_tunnel (struct GNUNET_MESH_Handle *h, MESH_TunnelNumber tid)
  * - Frees all memory used
  *
  * @param t Pointer to the tunnel.
+ * @param call_cleaner Whether to call the cleaner handler.
  *
  * @return Handle to the required tunnel or NULL if not found.
  */
 static void
-destroy_tunnel (struct GNUNET_MESH_Tunnel *t)
+destroy_tunnel (struct GNUNET_MESH_Tunnel *t, int call_cleaner)
 {
   struct GNUNET_MESH_Handle *h;
   struct GNUNET_PeerIdentity pi;
   struct GNUNET_MESH_TransmitHandle *th;
-  struct GNUNET_MESH_TransmitHandle *aux;
+  struct GNUNET_MESH_TransmitHandle *next;
   unsigned int i;
 
+  LOG (GNUNET_ERROR_TYPE_DEBUG, "destroy_tunnel %X\n", t->tid);
+
   if (NULL == t)
   {
     GNUNET_break (0);
     return;
   }
   h = t->mesh;
-  th = h->th_head;
-  while (NULL != th)
-  {
-    if (th->tunnel == t)
-    {
-      aux = th->next;
-      GNUNET_CONTAINER_DLL_remove (h->th_head, h->th_tail, th);
-      if (NULL == h->th_head && NULL != h->th)
-      {
-        GNUNET_CLIENT_notify_transmit_ready_cancel (h->th);
-        h->th = NULL;
-      }
-      if (NULL != th->notify)
-        th->notify (th->notify_cls, 0, NULL);
-      if (GNUNET_SCHEDULER_NO_TASK != th->timeout_task)
-        GNUNET_SCHEDULER_cancel (th->timeout_task);
-      GNUNET_free (th);
-      th = aux;
-    }
-    else
-    {
-      th = th->next;
-    }
-  }
+
+  /* disconnect all peers */
   GNUNET_CONTAINER_DLL_remove (h->tunnels_head, h->tunnels_tail, t);
   for (i = 0; i < t->npeers; i++)
   {
-    if (NULL != t->disconnect_handler && t->peers[i]->connected)
+    if ( (NULL != t->disconnect_handler) && t->peers[i]->connected)
     {
       GNUNET_PEER_resolve (t->peers[i]->id, &pi);
       t->disconnect_handler (t->cls, &pi);
@@ -423,10 +490,39 @@ destroy_tunnel (struct GNUNET_MESH_Tunnel *t)
     GNUNET_PEER_change_rc (t->peers[i]->id, -1);
     GNUNET_free (t->peers[i]);
   }
+
+  /* signal tunnel destruction */
+  if ( (NULL != h->cleaner) && (0 != t->owner) && (GNUNET_YES == call_cleaner) )
+    h->cleaner (h->cls, t, t->ctx);
+
+  /* check that clients did not leave messages behind in the queue */
+  for (th = h->th_head; NULL != th; th = next)
+  {
+    next = th->next;
+    if (th->tunnel != t)
+      continue;
+    /* Clients should have aborted their requests already.
+     * Management traffic should be ok, as clients can't cancel that */
+    GNUNET_break (GNUNET_NO == th_is_payload(th));
+    GNUNET_CONTAINER_DLL_remove (h->th_head, h->th_tail, th);
+
+    /* clean up request */
+    if (GNUNET_SCHEDULER_NO_TASK != th->timeout_task)
+      GNUNET_SCHEDULER_cancel (th->timeout_task);
+    GNUNET_free (th);    
+  }
+
+  /* if there are no more pending requests with mesh service, cancel active request */
+  /* Note: this should be unnecessary... */
+  if ((0 == message_ready_size (h)) && (NULL != h->th))
+  {
+    GNUNET_CLIENT_notify_transmit_ready_cancel (h->th);
+    h->th = NULL;
+  }
+
+
   if (t->npeers > 0)
     GNUNET_free (t->peers);
-  if (NULL != h->cleaner && 0 != t->owner)
-    h->cleaner (h->cls, t, t->ctx);
   if (0 != t->owner)
     GNUNET_PEER_change_rc (t->owner, -1);
   if (0 != t->napps && t->apps)
@@ -508,6 +604,7 @@ remove_peer_from_tunnel (struct GNUNET_MESH_Peer *p)
 
 /**
  * Notify client that the transmission has timed out
+ * 
  * @param cls closure
  * @param tc task context
  */
@@ -519,12 +616,13 @@ timeout_transmission (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
 
   mesh = th->tunnel->mesh;
   GNUNET_CONTAINER_DLL_remove (mesh->th_head, mesh->th_tail, th);
-  if (th->notify != NULL)
+  th->tunnel->packet_size = 0;
+  if (GNUNET_YES == th_is_payload (th))
     th->notify (th->notify_cls, 0, NULL);
   GNUNET_free (th);
-  if ((NULL == mesh->th_head) && (NULL != mesh->th))
+  if ((0 == message_ready_size (mesh)) && (NULL != mesh->th))
   {
-    /* queue empty, no point in asking for transmission */
+    /* nothing ready to transmit, no point in asking for transmission */
     GNUNET_CLIENT_notify_transmit_ready_cancel (mesh->th);
     mesh->th = NULL;
   }
@@ -545,7 +643,7 @@ add_to_queue (struct GNUNET_MESH_Handle *h,
   struct GNUNET_MESH_TransmitHandle *p;
 
   p = h->th_head;
-  while ((NULL != p) && (th->priority <= p->priority))
+  while ((NULL != p))
     p = p->next;
   if (NULL == p)
     p = h->th_tail;
@@ -564,12 +662,37 @@ add_to_queue (struct GNUNET_MESH_Handle *h,
  * Auxiliary function to send an already constructed packet to the service.
  * Takes care of creating a new queue element, copying the message and
  * calling the tmt_rdy function if necessary.
+ *
  * @param h mesh handle
  * @param msg message to transmit
+ * @param tunnel tunnel this send is related to (NULL if N/A)
  */
 static void
 send_packet (struct GNUNET_MESH_Handle *h,
-             const struct GNUNET_MessageHeader *msg);
+             const struct GNUNET_MessageHeader *msg,
+             struct GNUNET_MESH_Tunnel *tunnel);
+
+
+/**
+ * Send an ack on the tunnel to confirm the processing of a message.
+ * 
+ * @param h Mesh handle.
+ * @param t Tunnel on which to send the ACK.
+ */
+static void
+send_ack (struct GNUNET_MESH_Handle *h, struct GNUNET_MESH_Tunnel *t)
+{
+  struct GNUNET_MESH_LocalAck msg;
+
+  msg.header.type = htons (GNUNET_MESSAGE_TYPE_MESH_LOCAL_ACK);
+  msg.header.size = htons (sizeof (msg));
+  msg.tunnel_id = htonl (t->tid);
+  msg.max_pid = t->pid + 1;
+
+  send_packet (h, &msg.header, t);
+  return;
+}
+
 
 
 /**
@@ -598,7 +721,7 @@ send_connect (struct GNUNET_MESH_Handle *h)
   size += h->n_applications * sizeof (GNUNET_MESH_ApplicationType);
   size += h->n_handlers * sizeof (uint16_t);
   {
-    char buf[size];
+    char buf[size] GNUNET_ALIGN;
     struct GNUNET_MESH_ClientConnect *msg;
     GNUNET_MESH_ApplicationType *apps;
     uint16_t napps;
@@ -613,19 +736,17 @@ send_connect (struct GNUNET_MESH_Handle *h)
     for (napps = 0; napps < h->n_applications; napps++)
     {
       apps[napps] = htonl (h->applications[napps]);
-      LOG (GNUNET_ERROR_TYPE_DEBUG, "mesh:  app %u\n", h->applications[napps]);
+      LOG (GNUNET_ERROR_TYPE_DEBUG, " app %u\n", h->applications[napps]);
     }
     types = (uint16_t *) & apps[napps];
     for (ntypes = 0; ntypes < h->n_handlers; ntypes++)
       types[ntypes] = htons (h->message_handlers[ntypes].type);
     msg->applications = htons (napps);
     msg->types = htons (ntypes);
-#if MESH_API_DEBUG
     LOG (GNUNET_ERROR_TYPE_DEBUG,
-         "mesh: Sending %lu bytes long message %d types and %d apps\n",
+         "Sending %lu bytes long message %d types and %d apps\n",
          ntohs (msg->header.size), ntypes, napps);
-#endif
-    send_packet (h, &msg->header);
+    send_packet (h, &msg->header, NULL);
   }
 }
 
@@ -639,37 +760,39 @@ send_connect (struct GNUNET_MESH_Handle *h)
  * @return GNUNET_YES in case of sucess, GNUNET_NO otherwise (service down...)
  */
 static int
-reconnect (struct GNUNET_MESH_Handle *h)
+do_reconnect (struct GNUNET_MESH_Handle *h)
 {
   struct GNUNET_MESH_Tunnel *t;
   unsigned int i;
 
-#if MESH_API_DEBUG
-  LOG (GNUNET_ERROR_TYPE_DEBUG, "mesh: *****************************\n");
-  LOG (GNUNET_ERROR_TYPE_DEBUG, "mesh: *******   RECONNECT   *******\n");
-  LOG (GNUNET_ERROR_TYPE_DEBUG, "mesh: *****************************\n");
-#endif
-  h->in_receive = GNUNET_NO;
+  LOG (GNUNET_ERROR_TYPE_DEBUG, "*****************************\n");
+  LOG (GNUNET_ERROR_TYPE_DEBUG, "*******   RECONNECT   *******\n");
+  LOG (GNUNET_ERROR_TYPE_DEBUG, "*****************************\n");
+  LOG (GNUNET_ERROR_TYPE_DEBUG, "******** on %p *******\n", h);
+  LOG (GNUNET_ERROR_TYPE_DEBUG, "*****************************\n");
+
   /* disconnect */
   if (NULL != h->th)
   {
     GNUNET_CLIENT_notify_transmit_ready_cancel (h->th);
+    h->th = NULL;
   }
   if (NULL != h->client)
   {
-    GNUNET_CLIENT_disconnect (h->client, GNUNET_NO);
+    GNUNET_CLIENT_disconnect (h->client);
   }
 
   /* connect again */
   h->client = GNUNET_CLIENT_connect ("mesh", h->cfg);
   if (h->client == NULL)
   {
-    GNUNET_SCHEDULER_add_delayed (h->reconnect_time, &reconnect_cbk, h);
+    h->reconnect_task = GNUNET_SCHEDULER_add_delayed (h->reconnect_time,
+                                                      &reconnect_cbk, h);
     h->reconnect_time =
-        GNUNET_TIME_relative_min (GNUNET_TIME_UNIT_HOURS,
+        GNUNET_TIME_relative_min (GNUNET_TIME_UNIT_SECONDS,
                                   GNUNET_TIME_relative_multiply
                                   (h->reconnect_time, 2));
-    LOG (GNUNET_ERROR_TYPE_DEBUG, "mesh:   Next retry in %sms\n",
+    LOG (GNUNET_ERROR_TYPE_DEBUG, "  Next retry in %sms\n",
          GNUNET_TIME_relative_to_string (h->reconnect_time));
     GNUNET_break (0);
     return GNUNET_NO;
@@ -696,21 +819,20 @@ reconnect (struct GNUNET_MESH_Handle *h)
     tmsg.header.type = htons (GNUNET_MESSAGE_TYPE_MESH_LOCAL_TUNNEL_CREATE);
     tmsg.header.size = htons (sizeof (struct GNUNET_MESH_TunnelMessage));
     tmsg.tunnel_id = htonl (t->tid);
-    send_packet (h, &tmsg.header);
+    send_packet (h, &tmsg.header, t);
 
     pmsg.header.size = htons (sizeof (struct GNUNET_MESH_PeerControl));
     pmsg.header.type = htons (GNUNET_MESSAGE_TYPE_MESH_LOCAL_PEER_ADD);
     pmsg.tunnel_id = htonl (t->tid);
 
     /* Reconnect all peers */
-    for (i = 0; i < t->npeers; i++)
+    /* If the tunnel was "by type", dont connect individual peers */
+    for (i = 0; i < t->npeers && 0 == t->napps; i++)
     {
       GNUNET_PEER_resolve (t->peers[i]->id, &pmsg.peer);
       if (NULL != t->disconnect_handler && t->peers[i]->connected)
         t->disconnect_handler (t->cls, &pmsg.peer);
-      /* If the tunnel was "by type", dont connect individual peers */
-      if (0 == t->napps)
-        send_packet (t->mesh, &pmsg.header);
+      send_packet (t->mesh, &pmsg.header, t);
     }
     /* Reconnect all types, if any  */
     for (i = 0; i < t->napps; i++)
@@ -721,8 +843,12 @@ reconnect (struct GNUNET_MESH_Handle *h)
       msg.header.type = htons (GNUNET_MESSAGE_TYPE_MESH_LOCAL_PEER_ADD_BY_TYPE);
       msg.tunnel_id = htonl (t->tid);
       msg.type = htonl (t->apps[i]);
-      send_packet (t->mesh, &msg.header);
+      send_packet (t->mesh, &msg.header, t);
     }
+    if (GNUNET_NO == t->buffering)
+      GNUNET_MESH_tunnel_buffer (t, GNUNET_NO);
+    if (GNUNET_YES == t->speed_min)
+      GNUNET_MESH_tunnel_speed_min (t);
   }
   return GNUNET_YES;
 }
@@ -738,9 +864,29 @@ reconnect_cbk (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
 {
   struct GNUNET_MESH_Handle *h = cls;
 
-  if (tc->reason == GNUNET_SCHEDULER_REASON_SHUTDOWN)
+  h->reconnect_task = GNUNET_SCHEDULER_NO_TASK;
+  if (0 != (tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN))
     return;
-  reconnect (h);
+  do_reconnect (h);
+}
+
+
+/**
+ * Reconnect to the service, retransmit all infomation to try to restore the
+ * original state.
+ *
+ * @param h handle to the mesh
+ *
+ * @return GNUNET_YES in case of sucess, GNUNET_NO otherwise (service down...)
+ */
+static void
+reconnect (struct GNUNET_MESH_Handle *h)
+{
+  LOG (GNUNET_ERROR_TYPE_DEBUG, "Requested RECONNECT\n");
+  h->in_receive = GNUNET_NO;
+  if (GNUNET_SCHEDULER_NO_TASK == h->reconnect_task)
+    h->reconnect_task = GNUNET_SCHEDULER_add_delayed (h->reconnect_time,
+                                                      &reconnect_cbk, h);
 }
 
 
@@ -759,7 +905,6 @@ process_tunnel_created (struct GNUNET_MESH_Handle *h,
                         const struct GNUNET_MESH_TunnelNotification *msg)
 {
   struct GNUNET_MESH_Tunnel *t;
-  struct GNUNET_ATS_Information atsi;
   MESH_TunnelNumber tid;
 
   tid = ntohl (msg->tunnel_id);
@@ -768,21 +913,37 @@ process_tunnel_created (struct GNUNET_MESH_Handle *h,
     GNUNET_break (0);
     return;
   }
-  t = create_tunnel (h, tid);
-  t->owner = GNUNET_PEER_intern (&msg->peer);
-  t->npeers = 1;
-  t->peers = GNUNET_malloc (sizeof (struct GNUNET_MESH_Peer *));
-  t->peers[0] = GNUNET_malloc (sizeof (struct GNUNET_MESH_Peer));
-  t->peers[0]->t = t;
-  t->peers[0]->connected = 1;
-  t->peers[0]->id = t->owner;
-  t->mesh = h;
-  t->tid = tid;
   if (NULL != h->new_tunnel)
   {
+    struct GNUNET_ATS_Information atsi;
+
+    t = create_tunnel (h, tid);
+    t->owner = GNUNET_PEER_intern (&msg->peer);
+    t->npeers = 1;
+    t->peers = GNUNET_malloc (sizeof (struct GNUNET_MESH_Peer *));
+    t->peers[0] = GNUNET_malloc (sizeof (struct GNUNET_MESH_Peer));
+    t->peers[0]->t = t;
+    t->peers[0]->connected = 1;
+    t->peers[0]->id = t->owner;
+    GNUNET_PEER_change_rc (t->owner, 1);
+    t->mesh = h;
+    t->tid = tid;
     atsi.type = 0;
     atsi.value = 0;
     t->ctx = h->new_tunnel (h->cls, t, &msg->peer, &atsi);
+    LOG (GNUNET_ERROR_TYPE_DEBUG, "new incoming tunnel %X\n", t->tid);
+  }
+  else
+  {
+    struct GNUNET_MESH_TunnelMessage d_msg;
+
+    LOG (GNUNET_ERROR_TYPE_DEBUG, "No handler for incoming tunnels\n");
+
+    d_msg.header.type = htons (GNUNET_MESSAGE_TYPE_MESH_LOCAL_TUNNEL_DESTROY);
+    d_msg.header.size = htons (sizeof (struct GNUNET_MESH_TunnelMessage));
+    d_msg.tunnel_id = msg->tunnel_id;
+
+    send_packet (h, &d_msg.header, NULL);
   }
   return;
 }
@@ -812,8 +973,8 @@ process_tunnel_destroy (struct GNUNET_MESH_Handle *h,
   {
     GNUNET_break (0);
   }
-
-  destroy_tunnel (t);
+  LOG (GNUNET_ERROR_TYPE_DEBUG, "tunnel %X destroyed\n", t->tid);
+  destroy_tunnel (t, GNUNET_YES);
   return;
 }
 
@@ -834,7 +995,7 @@ process_peer_event (struct GNUNET_MESH_Handle *h,
   GNUNET_PEER_Id id;
   uint16_t size;
 
-  LOG (GNUNET_ERROR_TYPE_DEBUG, "mesh: processig peer event\n");
+  LOG (GNUNET_ERROR_TYPE_DEBUG, "processig peer event\n");
   size = ntohs (msg->header.size);
   if (size != sizeof (struct GNUNET_MESH_PeerControl))
   {
@@ -852,7 +1013,7 @@ process_peer_event (struct GNUNET_MESH_Handle *h,
     p = add_peer_to_tunnel (t, &msg->peer);
   if (GNUNET_MESSAGE_TYPE_MESH_LOCAL_PEER_ADD == ntohs (msg->header.type))
   {
-    LOG (GNUNET_ERROR_TYPE_DEBUG, "mesh: adding peer\n");
+    LOG (GNUNET_ERROR_TYPE_DEBUG, "adding peer\n");
     if (NULL != t->connect_handler)
     {
       atsi.type = 0;
@@ -863,7 +1024,7 @@ process_peer_event (struct GNUNET_MESH_Handle *h,
   }
   else
   {
-    LOG (GNUNET_ERROR_TYPE_DEBUG, "mesh: removing peer\n");
+    LOG (GNUNET_ERROR_TYPE_DEBUG, "removing peer\n");
     if (NULL != t->disconnect_handler && p->connected)
     {
       t->disconnect_handler (t->cls, &msg->peer);
@@ -871,7 +1032,7 @@ process_peer_event (struct GNUNET_MESH_Handle *h,
     remove_peer_from_tunnel (p);
     GNUNET_free (p);
   }
-  LOG (GNUNET_ERROR_TYPE_DEBUG, "mesh: processing peer event END\n");
+  LOG (GNUNET_ERROR_TYPE_DEBUG, "processing peer event END\n");
 }
 
 
@@ -880,8 +1041,11 @@ process_peer_event (struct GNUNET_MESH_Handle *h,
  *
  * @param h         The mesh handle
  * @param message   A message encapsulating the data
+ * 
+ * @return GNUNET_YES if everything went fine
+ *         GNUNET_NO if client closed connection (h no longer valid)
  */
-static void
+static int
 process_incoming_data (struct GNUNET_MESH_Handle *h,
                        const struct GNUNET_MessageHeader *message)
 {
@@ -895,7 +1059,7 @@ process_incoming_data (struct GNUNET_MESH_Handle *h,
   unsigned int i;
   uint16_t type;
 
-  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "mesh: Got a data message!\n");
+  LOG (GNUNET_ERROR_TYPE_DEBUG, "Got a data message!\n");
   type = ntohs (message->type);
   switch (type)
   {
@@ -905,29 +1069,33 @@ process_incoming_data (struct GNUNET_MESH_Handle *h,
     t = retrieve_tunnel (h, ntohl (ucast->tid));
     payload = (struct GNUNET_MessageHeader *) &ucast[1];
     peer = &ucast->oid;
-    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "mesh:   on tunnel %s [%x]\n",
-                GNUNET_i2s (peer), ntohl (ucast->tid));
+    LOG (GNUNET_ERROR_TYPE_DEBUG, "  ucast on tunnel %s [%X]\n",
+         GNUNET_i2s (peer), ntohl (ucast->tid));
     break;
   case GNUNET_MESSAGE_TYPE_MESH_MULTICAST:
     mcast = (struct GNUNET_MESH_Multicast *) message;
     t = retrieve_tunnel (h, ntohl (mcast->tid));
     payload = (struct GNUNET_MessageHeader *) &mcast[1];
     peer = &mcast->oid;
+    LOG (GNUNET_ERROR_TYPE_DEBUG, "  mcast on tunnel %s [%X]\n",
+         GNUNET_i2s (peer), ntohl (mcast->tid));
     break;
   case GNUNET_MESSAGE_TYPE_MESH_TO_ORIGIN:
     to_orig = (struct GNUNET_MESH_ToOrigin *) message;
     t = retrieve_tunnel (h, ntohl (to_orig->tid));
     payload = (struct GNUNET_MessageHeader *) &to_orig[1];
     peer = &to_orig->sender;
+    LOG (GNUNET_ERROR_TYPE_DEBUG, "  torig on tunnel %s [%X]\n",
+         GNUNET_i2s (peer), ntohl (to_orig->tid));
     break;
   default:
     GNUNET_break (0);
-    return;
+    return GNUNET_YES;
   }
   if (NULL == t)
   {
-    GNUNET_break (0);
-    return;
+    /* Tunnel was ignored, probably service didn't get it yet */
+    return GNUNET_YES;
   }
   type = ntohs (payload->type);
   for (i = 0; i < h->n_handlers; i++)
@@ -942,20 +1110,63 @@ process_incoming_data (struct GNUNET_MESH_Handle *h,
       if (GNUNET_OK !=
           handler->callback (h->cls, t, &t->ctx, peer, payload, &atsi))
       {
-        LOG (GNUNET_ERROR_TYPE_DEBUG, "MESH: callback caused disconnection\n");
+        LOG (GNUNET_ERROR_TYPE_DEBUG, "callback caused disconnection\n");
         GNUNET_MESH_disconnect (h);
-        return;
+        return GNUNET_NO;
       }
-#if MESH_API_DEBUG
       else
       {
         LOG (GNUNET_ERROR_TYPE_DEBUG,
-             "MESH: callback completed successfully\n");
-
+             "callback completed successfully\n");
+        send_ack (h, t);
       }
-#endif
     }
   }
+  return GNUNET_YES;
+}
+
+
+/**
+ * Process a local ACK message, enabling the client to send
+ * more data to the service.
+ * 
+ * @param h Mesh handle.
+ * @param message Message itself.
+ */
+static void
+process_ack (struct GNUNET_MESH_Handle *h,
+             const struct GNUNET_MessageHeader *message)
+{
+  struct GNUNET_MESH_LocalAck *msg;
+  struct GNUNET_MESH_Tunnel *t;
+  uint32_t ack;
+
+  LOG (GNUNET_ERROR_TYPE_DEBUG, "Got an ACK!\n");
+  msg = (struct GNUNET_MESH_LocalAck *) message;
+
+  t = retrieve_tunnel (h, ntohl (msg->tunnel_id));
+
+  if (NULL == t)
+  {
+    LOG (GNUNET_ERROR_TYPE_WARNING,
+         "ACK on unknown tunnel %X\n",
+         ntohl (msg->tunnel_id));
+    return;
+  }
+  ack = ntohl (msg->max_pid);
+  LOG (GNUNET_ERROR_TYPE_DEBUG, "  on tunnel %X, ack %u!\n", t->tid, ack);
+  if (ack > t->max_pid || PID_OVERFLOW (t->max_pid, ack))
+    t->max_pid = ack;
+  else
+    return;
+  if (NULL == h->th && 0 < t->packet_size)
+  {
+    LOG (GNUNET_ERROR_TYPE_DEBUG, "  tmt rdy was NULL, requesting!\n", t->tid, ack);
+    h->th =
+        GNUNET_CLIENT_notify_transmit_ready (h->client, t->packet_size,
+                                             GNUNET_TIME_UNIT_FOREVER_REL,
+                                             GNUNET_YES, &send_callback, h);
+  }
 }
 
 
@@ -972,11 +1183,12 @@ msg_received (void *cls, const struct GNUNET_MessageHeader *msg)
 
   if (msg == NULL)
   {
+    LOG (GNUNET_ERROR_TYPE_WARNING, "Received NULL msg on %p\n", h);
     reconnect (h);
     return;
   }
-  LOG (GNUNET_ERROR_TYPE_DEBUG, "mesh: received a message type %hu from MESH\n",
-       ntohs (msg->type));
+  LOG (GNUNET_ERROR_TYPE_DEBUG, "received a message type %s from MESH\n",
+       GNUNET_MESH_DEBUG_M2S (ntohs (msg->type)));
   switch (ntohs (msg->type))
   {
     /* Notify of a new incoming tunnel */
@@ -996,17 +1208,29 @@ msg_received (void *cls, const struct GNUNET_MessageHeader *msg)
   case GNUNET_MESSAGE_TYPE_MESH_UNICAST:
   case GNUNET_MESSAGE_TYPE_MESH_MULTICAST:
   case GNUNET_MESSAGE_TYPE_MESH_TO_ORIGIN:
-    process_incoming_data (h, msg);
+    if (GNUNET_NO == process_incoming_data (h, msg))
+      return;
+    break;
+  case GNUNET_MESSAGE_TYPE_MESH_LOCAL_ACK:
+    process_ack (h, msg);
     break;
-    /* We shouldn't get any other packages, log and ignore */
   default:
+    /* We shouldn't get any other packages, log and ignore */
     LOG (GNUNET_ERROR_TYPE_WARNING,
-         "MESH: unsolicited message form service (type %d)\n",
+         "unsolicited message form service (type %hu)\n",
          ntohs (msg->type));
   }
-  LOG (GNUNET_ERROR_TYPE_DEBUG, "mesh: message processed\n");
-  GNUNET_CLIENT_receive (h->client, &msg_received, h,
-                         GNUNET_TIME_UNIT_FOREVER_REL);
+  LOG (GNUNET_ERROR_TYPE_DEBUG, "message processed\n");
+  if (GNUNET_YES == h->in_receive)
+  {
+    GNUNET_CLIENT_receive (h->client, &msg_received, h,
+                           GNUNET_TIME_UNIT_FOREVER_REL);
+  }
+  else
+  {
+    LOG (GNUNET_ERROR_TYPE_DEBUG,
+         "in receive off, not calling CLIENT_receive\n");
+  }
 }
 
 
@@ -1029,23 +1253,36 @@ send_callback (void *cls, size_t size, void *buf)
 {
   struct GNUNET_MESH_Handle *h = cls;
   struct GNUNET_MESH_TransmitHandle *th;
+  struct GNUNET_MESH_TransmitHandle *next;
+  struct GNUNET_MESH_Tunnel *t;
   char *cbuf = buf;
   size_t tsize;
   size_t psize;
 
-  LOG (GNUNET_ERROR_TYPE_DEBUG, "mesh: Send packet() Buffer %u\n", size);
-  h->th = NULL;
+  LOG (GNUNET_ERROR_TYPE_DEBUG, "Send packet() Buffer %u\n", size);
   if ((0 == size) || (NULL == buf))
   {
+    LOG (GNUNET_ERROR_TYPE_DEBUG, "Received NULL send callback on %p\n", h);
     reconnect (h);
+    h->th = NULL;
     return 0;
   }
   tsize = 0;
-  while ((NULL != (th = h->th_head)) && (size >= th->size))
+  next = h->th_head;
+  while ((NULL != (th = next)) && (size >= th->size))
   {
-    if (NULL != th->notify)
+    t = th->tunnel;
+    if (GNUNET_YES == th_is_payload (th))
     {
-      if (th->tunnel->tid >= GNUNET_MESH_LOCAL_TUNNEL_ID_SERV)
+      LOG (GNUNET_ERROR_TYPE_DEBUG, " payload\n");
+      if (t->max_pid < t->pid && GNUNET_NO == PID_OVERFLOW (t->pid, t->max_pid))
+      {
+        /* This tunnel is not ready to transmit yet, try next message */
+        next = th->next;
+        continue;
+      }
+      t->packet_size = 0;
+      if (t->tid >= GNUNET_MESH_LOCAL_TUNNEL_ID_SERV)
       {
         /* traffic to origin */
         struct GNUNET_MESH_ToOrigin to;
@@ -1054,15 +1291,16 @@ send_callback (void *cls, size_t size, void *buf)
         GNUNET_assert (size >= th->size);
         mh = (struct GNUNET_MessageHeader *) &cbuf[sizeof (to)];
         psize = th->notify (th->notify_cls, size - sizeof (to), mh);
-        GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "mesh:   to origin, type %u\n",
-                    ntohs (mh->type));
+        LOG (GNUNET_ERROR_TYPE_DEBUG, "  to origin, type %s\n",
+             GNUNET_MESH_DEBUG_M2S (ntohs (mh->type)));
         if (psize > 0)
         {
           psize += sizeof (to);
           GNUNET_assert (size >= psize);
           to.header.size = htons (psize);
           to.header.type = htons (GNUNET_MESSAGE_TYPE_MESH_TO_ORIGIN);
-          to.tid = htonl (th->tunnel->tid);
+          to.tid = htonl (t->tid);
+          // FIXME pid?
           memset (&to.oid, 0, sizeof (struct GNUNET_PeerIdentity));
           memset (&to.sender, 0, sizeof (struct GNUNET_PeerIdentity));
           memcpy (cbuf, &to, sizeof (to));
@@ -1077,16 +1315,17 @@ send_callback (void *cls, size_t size, void *buf)
         GNUNET_assert (size >= th->size);
         mh = (struct GNUNET_MessageHeader *) &cbuf[sizeof (mc)];
         psize = th->notify (th->notify_cls, size - sizeof (mc), mh);
-        GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "mesh:   multicast, type %u\n",
-                    ntohs (mh->type));
+        LOG (GNUNET_ERROR_TYPE_DEBUG, "  multicast, type %s\n",
+             GNUNET_MESH_DEBUG_M2S (ntohs (mh->type)));
         if (psize > 0)
         {
           psize += sizeof (mc);
           GNUNET_assert (size >= psize);
           mc.header.size = htons (psize);
           mc.header.type = htons (GNUNET_MESSAGE_TYPE_MESH_MULTICAST);
-          mc.tid = htonl (th->tunnel->tid);
-          mc.mid = 0;
+          mc.tid = htonl (t->tid);
+          mc.pid = htonl (t->pid);
+          mc.ttl = 0;
           memset (&mc.oid, 0, sizeof (struct GNUNET_PeerIdentity));
           memcpy (cbuf, &mc, sizeof (mc));
         }
@@ -1100,23 +1339,29 @@ send_callback (void *cls, size_t size, void *buf)
         GNUNET_assert (size >= th->size);
         mh = (struct GNUNET_MessageHeader *) &cbuf[sizeof (uc)];
         psize = th->notify (th->notify_cls, size - sizeof (uc), mh);
-        GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "mesh:   unicast, type %u\n",
-                    ntohs (mh->type));
+        LOG (GNUNET_ERROR_TYPE_DEBUG, "  unicast, type %s\n",
+             GNUNET_MESH_DEBUG_M2S (ntohs (mh->type)));
         if (psize > 0)
         {
           psize += sizeof (uc);
           GNUNET_assert (size >= psize);
           uc.header.size = htons (psize);
           uc.header.type = htons (GNUNET_MESSAGE_TYPE_MESH_UNICAST);
-          uc.tid = htonl (th->tunnel->tid);
+          uc.tid = htonl (t->tid);
+          uc.pid = htonl (t->pid);
           memset (&uc.oid, 0, sizeof (struct GNUNET_PeerIdentity));
           GNUNET_PEER_resolve (th->target, &uc.destination);
           memcpy (cbuf, &uc, sizeof (uc));
         }
       }
+      t->pid++;
     }
     else
     {
+      struct GNUNET_MessageHeader *mh = (struct GNUNET_MessageHeader *) &th[1];
+
+      LOG (GNUNET_ERROR_TYPE_DEBUG, "  mesh traffic, type %s\n",
+             GNUNET_MESH_DEBUG_M2S (ntohs (mh->type)));
       memcpy (cbuf, &th[1], th->size);
       psize = th->size;
     }
@@ -1124,26 +1369,34 @@ send_callback (void *cls, size_t size, void *buf)
       GNUNET_SCHEDULER_cancel (th->timeout_task);
     GNUNET_CONTAINER_DLL_remove (h->th_head, h->th_tail, th);
     GNUNET_free (th);
+    next = h->th_head;
     cbuf += psize;
     size -= psize;
     tsize += psize;
   }
-  LOG (GNUNET_ERROR_TYPE_DEBUG, "mesh:   total size: %u\n", tsize);
-  if (NULL != (th = h->th_head))
+  LOG (GNUNET_ERROR_TYPE_DEBUG, "  total size: %u\n", tsize);
+  h->th = NULL;
+  size = message_ready_size (h);
+  if (0 != size)
   {
-    LOG (GNUNET_ERROR_TYPE_DEBUG, "mesh:   next size: %u\n", th->size);
+    LOG (GNUNET_ERROR_TYPE_DEBUG, "  next size: %u\n", size);
     h->th =
-        GNUNET_CLIENT_notify_transmit_ready (h->client, th->size,
+        GNUNET_CLIENT_notify_transmit_ready (h->client, size,
                                              GNUNET_TIME_UNIT_FOREVER_REL,
                                              GNUNET_YES, &send_callback, h);
   }
-  LOG (GNUNET_ERROR_TYPE_DEBUG, "mesh: Send packet() END\n");
+  else
+  {
+    LOG (GNUNET_ERROR_TYPE_DEBUG, "  nothing left to transmit\n");
+  }
   if (GNUNET_NO == h->in_receive)
   {
+    LOG (GNUNET_ERROR_TYPE_DEBUG, " start receiving from service\n");
     h->in_receive = GNUNET_YES;
     GNUNET_CLIENT_receive (h->client, &msg_received, h,
                            GNUNET_TIME_UNIT_FOREVER_REL);
   }
+  LOG (GNUNET_ERROR_TYPE_DEBUG, "Send packet() END\n");
   return tsize;
 }
 
@@ -1152,21 +1405,24 @@ send_callback (void *cls, size_t size, void *buf)
  * Auxiliary function to send an already constructed packet to the service.
  * Takes care of creating a new queue element, copying the message and
  * calling the tmt_rdy function if necessary.
+ * 
  * @param h mesh handle
  * @param msg message to transmit
+ * @param tunnel tunnel this send is related to (NULL if N/A)
  */
 static void
 send_packet (struct GNUNET_MESH_Handle *h,
-             const struct GNUNET_MessageHeader *msg)
+             const struct GNUNET_MessageHeader *msg,
+             struct GNUNET_MESH_Tunnel *tunnel)
 {
   struct GNUNET_MESH_TransmitHandle *th;
   size_t msize;
 
   msize = ntohs (msg->size);
   th = GNUNET_malloc (sizeof (struct GNUNET_MESH_TransmitHandle) + msize);
-  th->priority = UINT32_MAX;
   th->timeout = GNUNET_TIME_UNIT_FOREVER_ABS;
   th->size = msize;
+  th->tunnel = tunnel;
   memcpy (&th[1], msg, msize);
   add_to_queue (h, th);
   if (NULL != h->th)
@@ -1186,13 +1442,12 @@ send_packet (struct GNUNET_MESH_Handle *h,
  * Connect to the mesh service.
  *
  * @param cfg configuration to use
- * @param queue_size size of the data message queue, shared among all tunnels
- *                   (each tunnel is guaranteed to accept at least one message,
- *                    no matter what is the status of other tunnels)
  * @param cls closure for the various callbacks that follow
  *            (including handlers in the handlers array)
  * @param new_tunnel function called when an *inbound* tunnel is created
- * @param cleaner function called when an *inbound* tunnel is destroyed
+ * @param cleaner function called when an *inbound* tunnel is destroyed by the
+ *                remote peer, it is *not* called if GNUNET_MESH_tunnel_destroy
+ *                is called on the tunnel
  * @param handlers callbacks for messages we care about, NULL-terminated
  *                note that the mesh is allowed to drop notifications about
  *                inbound messages if the client does not process them fast
@@ -1202,8 +1457,7 @@ send_packet (struct GNUNET_MESH_Handle *h,
  *         (in this case, init is never called)
  */
 struct GNUNET_MESH_Handle *
-GNUNET_MESH_connect (const struct GNUNET_CONFIGURATION_Handle *cfg,
-                     unsigned int queue_size, void *cls,
+GNUNET_MESH_connect (const struct GNUNET_CONFIGURATION_Handle *cfg, void *cls,
                      GNUNET_MESH_InboundTunnelNotificationHandler new_tunnel,
                      GNUNET_MESH_TunnelEndHandler cleaner,
                      const struct GNUNET_MESH_MessageHandler *handlers,
@@ -1211,10 +1465,10 @@ GNUNET_MESH_connect (const struct GNUNET_CONFIGURATION_Handle *cfg,
 {
   struct GNUNET_MESH_Handle *h;
 
-  LOG (GNUNET_ERROR_TYPE_DEBUG, "mesh: GNUNET_MESH_connect()\n");
+  LOG (GNUNET_ERROR_TYPE_DEBUG, "GNUNET_MESH_connect()\n");
   h = GNUNET_malloc (sizeof (struct GNUNET_MESH_Handle));
+  LOG (GNUNET_ERROR_TYPE_DEBUG, " addr %p\n", h);
   h->cfg = cfg;
-  h->max_queue_size = queue_size;
   h->new_tunnel = new_tunnel;
   h->cleaner = cleaner;
   h->client = GNUNET_CLIENT_connect ("mesh", cfg);
@@ -1230,18 +1484,22 @@ GNUNET_MESH_connect (const struct GNUNET_CONFIGURATION_Handle *cfg,
   h->message_handlers = handlers;
   h->next_tid = GNUNET_MESH_LOCAL_TUNNEL_ID_CLI;
   h->reconnect_time = GNUNET_TIME_UNIT_MILLISECONDS;
+  h->reconnect_task = GNUNET_SCHEDULER_NO_TASK;
 
   /* count handlers and apps, calculate size */
   for (h->n_applications = 0; stypes[h->n_applications]; h->n_applications++) ;
   for (h->n_handlers = 0; handlers[h->n_handlers].type; h->n_handlers++) ;
   send_connect (h);
-  LOG (GNUNET_ERROR_TYPE_DEBUG, "mesh: GNUNET_MESH_connect() END\n");
+  LOG (GNUNET_ERROR_TYPE_DEBUG, "GNUNET_MESH_connect() END\n");
   return h;
 }
 
 
 /**
- * Disconnect from the mesh service.
+ * Disconnect from the mesh service. All tunnels will be destroyed. All tunnel
+ * disconnect callbacks will be called on any still connected peers, notifying
+ * about their disconnection. The registered inbound tunnel cleaner will be
+ * called should any inbound tunnels still exist.
  *
  * @param handle connection to mesh to disconnect
  */
@@ -1250,26 +1508,100 @@ GNUNET_MESH_disconnect (struct GNUNET_MESH_Handle *handle)
 {
   struct GNUNET_MESH_Tunnel *t;
   struct GNUNET_MESH_Tunnel *aux;
+  struct GNUNET_MESH_TransmitHandle *th;
+
+  LOG (GNUNET_ERROR_TYPE_DEBUG, "MESH DISCONNECT\n");
 
   t = handle->tunnels_head;
   while (NULL != t)
   {
     aux = t->next;
-    destroy_tunnel (t);
+    if (t->tid < GNUNET_MESH_LOCAL_TUNNEL_ID_SERV)
+    {
+      GNUNET_break (0);
+      LOG (GNUNET_ERROR_TYPE_DEBUG, "tunnel %X not destroyed\n", t->tid);
+    }
+    destroy_tunnel (t, GNUNET_YES);
     t = aux;
   }
+  while ( (th = handle->th_head) != NULL)
+  {
+    struct GNUNET_MessageHeader *msg;
+
+    /* Make sure it is an allowed packet (everything else should have been
+     * already canceled).
+     */
+    GNUNET_break (GNUNET_NO == th_is_payload (th));
+    msg = (struct GNUNET_MessageHeader *) &th[1];
+    switch (ntohs(msg->type))
+    {
+      case GNUNET_MESSAGE_TYPE_MESH_LOCAL_CONNECT:
+      case GNUNET_MESSAGE_TYPE_MESH_LOCAL_TUNNEL_DESTROY:
+        break;
+      default:
+        GNUNET_break (0);
+        LOG (GNUNET_ERROR_TYPE_DEBUG, "unexpected msg %u\n",
+             ntohs(msg->type));
+    }
+
+    GNUNET_CONTAINER_DLL_remove (handle->th_head, handle->th_tail, th);
+    GNUNET_free (th);
+  }
+
   if (NULL != handle->th)
   {
     GNUNET_CLIENT_notify_transmit_ready_cancel (handle->th);
+    handle->th = NULL;
   }
   if (NULL != handle->client)
   {
-    GNUNET_CLIENT_disconnect (handle->client, GNUNET_NO);
+    GNUNET_CLIENT_disconnect (handle->client);
+    handle->client = NULL;
+  }
+  if (GNUNET_SCHEDULER_NO_TASK != handle->reconnect_task)
+  {
+    GNUNET_SCHEDULER_cancel(handle->reconnect_task);
+    handle->reconnect_task = GNUNET_SCHEDULER_NO_TASK;
   }
   GNUNET_free (handle);
 }
 
 
+/**
+ * Announce to ther peer the availability of services described by the regex,
+ * in order to be reachable to other peers via connect_by_string.
+ * 
+ * Note that the first 8 characters are considered to be part of a prefix,
+ * (for instance 'gnunet://'). If you put a variable part in there (*, +. ()),
+ * all matching strings will be stored in the DHT.
+ *
+ * @param h handle to mesh.
+ * @param regex string with the regular expression describing local services.
+ */
+void
+GNUNET_MESH_announce_regex (struct GNUNET_MESH_Handle *h,
+                            const char *regex)
+{
+  struct GNUNET_MessageHeader *msg;
+  size_t len;
+  size_t msgsize;
+
+  len = strlen (regex);
+  msgsize = sizeof(struct GNUNET_MessageHeader) + len;
+  GNUNET_assert (UINT16_MAX > msgsize);
+
+  {
+    char buffer[msgsize];
+
+    msg = (struct GNUNET_MessageHeader *) buffer;
+    msg->size = htons (msgsize);
+    msg->type = htons (GNUNET_MESSAGE_TYPE_MESH_LOCAL_ANNOUNCE_REGEX);
+    memcpy (&msg[1], regex, len);
+
+    send_packet(h, msg, NULL);
+  }
+}
+
 /**
  * Create a new tunnel (we're initiator and will be allowed to add/remove peers
  * and to broadcast).
@@ -1289,7 +1621,7 @@ GNUNET_MESH_tunnel_create (struct GNUNET_MESH_Handle *h, void *tunnel_ctx,
   struct GNUNET_MESH_Tunnel *t;
   struct GNUNET_MESH_TunnelMessage msg;
 
-  LOG (GNUNET_ERROR_TYPE_DEBUG, "mesh: Creating new tunnel\n");
+  LOG (GNUNET_ERROR_TYPE_DEBUG, "Creating new tunnel\n");
   t = create_tunnel (h, 0);
   t->connect_handler = connect_handler;
   t->disconnect_handler = disconnect_handler;
@@ -1298,13 +1630,14 @@ GNUNET_MESH_tunnel_create (struct GNUNET_MESH_Handle *h, void *tunnel_ctx,
   msg.header.type = htons (GNUNET_MESSAGE_TYPE_MESH_LOCAL_TUNNEL_CREATE);
   msg.header.size = htons (sizeof (struct GNUNET_MESH_TunnelMessage));
   msg.tunnel_id = htonl (t->tid);
-  send_packet (h, &msg.header);
+  send_packet (h, &msg.header, t);
   return t;
 }
 
 
 /**
- * Destroy an existing tunnel.
+ * Destroy an existing tunnel. The existing callback for the tunnel will NOT
+ * be called.
  *
  * @param tunnel tunnel handle
  */
@@ -1313,20 +1646,106 @@ GNUNET_MESH_tunnel_destroy (struct GNUNET_MESH_Tunnel *tunnel)
 {
   struct GNUNET_MESH_Handle *h;
   struct GNUNET_MESH_TunnelMessage msg;
+  struct GNUNET_MESH_TransmitHandle *th;
 
-  LOG (GNUNET_ERROR_TYPE_DEBUG, "mesh: Destroying tunnel\n");
+  LOG (GNUNET_ERROR_TYPE_DEBUG, "Destroying tunnel\n");
   h = tunnel->mesh;
 
-  if (0 != tunnel->owner)
-    GNUNET_PEER_change_rc (tunnel->owner, -1);
-
   msg.header.type = htons (GNUNET_MESSAGE_TYPE_MESH_LOCAL_TUNNEL_DESTROY);
   msg.header.size = htons (sizeof (struct GNUNET_MESH_TunnelMessage));
   msg.tunnel_id = htonl (tunnel->tid);
-  destroy_tunnel (tunnel);
-  send_packet (h, &msg.header);
+  th = h->th_head;
+  while (th != NULL)
+  {
+    struct GNUNET_MESH_TransmitHandle *aux;
+    if (th->tunnel == tunnel)
+    {
+      aux = th->next;
+      /* FIXME call the handler? */
+      if (GNUNET_YES == th_is_payload (th))
+        th->notify (th->notify_cls, 0, NULL);
+      GNUNET_CONTAINER_DLL_remove (h->th_head, h->th_tail, th);
+      GNUNET_free (th);
+      th = aux;
+    }
+    else
+      th = th->next;
+  }
+
+  destroy_tunnel (tunnel, GNUNET_NO);
+  send_packet (h, &msg.header, NULL);
+}
+
+/**
+ * Request that the tunnel data rate is limited to the speed of the slowest
+ * receiver.
+ *
+ * @param tunnel Tunnel affected.
+ */
+void
+GNUNET_MESH_tunnel_speed_min (struct GNUNET_MESH_Tunnel *tunnel)
+{
+  struct GNUNET_MESH_TunnelMessage msg;
+  struct GNUNET_MESH_Handle *h;
+
+  h = tunnel->mesh;
+  tunnel->speed_min = GNUNET_YES;
+
+  msg.header.type = htons (GNUNET_MESSAGE_TYPE_MESH_LOCAL_TUNNEL_MIN);
+  msg.header.size = htons (sizeof (struct GNUNET_MESH_TunnelMessage));
+  msg.tunnel_id = htonl (tunnel->tid);
+
+  send_packet (h, &msg.header, NULL);
+}
+
+
+/**
+ * Request that the tunnel data rate is limited to the speed of the fastest
+ * receiver. This is the default behavior.
+ *
+ * @param tunnel Tunnel affected.
+ */
+void
+GNUNET_MESH_tunnel_speed_max (struct GNUNET_MESH_Tunnel *tunnel)
+{
+  struct GNUNET_MESH_TunnelMessage msg;
+  struct GNUNET_MESH_Handle *h;
+
+  h = tunnel->mesh;
+  tunnel->speed_min = GNUNET_NO;
+
+  msg.header.type = htons (GNUNET_MESSAGE_TYPE_MESH_LOCAL_TUNNEL_MAX);
+  msg.header.size = htons (sizeof (struct GNUNET_MESH_TunnelMessage));
+  msg.tunnel_id = htonl (tunnel->tid);
+
+  send_packet (h, &msg.header, NULL);
 }
 
+/**
+ * Turn on/off the buffering status of the tunnel.
+ * 
+ * @param tunnel Tunnel affected.
+ * @param buffer GNUNET_YES to turn buffering on (default),
+ *               GNUNET_NO otherwise.
+ */
+void
+GNUNET_MESH_tunnel_buffer (struct GNUNET_MESH_Tunnel *tunnel, int buffer)
+{
+  struct GNUNET_MESH_TunnelMessage msg;
+  struct GNUNET_MESH_Handle *h;
+
+  h = tunnel->mesh;
+  tunnel->buffering = buffer;
+
+  if (GNUNET_YES == buffer)
+    msg.header.type = htons (GNUNET_MESSAGE_TYPE_MESH_LOCAL_TUNNEL_BUFFER);
+  else
+    msg.header.type = htons (GNUNET_MESSAGE_TYPE_MESH_LOCAL_TUNNEL_NOBUFFER);
+  msg.header.size = htons (sizeof (struct GNUNET_MESH_TunnelMessage));
+  msg.tunnel_id = htonl (tunnel->tid);
+
+  send_packet (h, &msg.header, NULL);
+}
 
 /**
  * Request that a peer should be added to the tunnel.  The existing
@@ -1363,7 +1782,7 @@ GNUNET_MESH_peer_request_connect_add (struct GNUNET_MESH_Tunnel *tunnel,
   msg.header.type = htons (GNUNET_MESSAGE_TYPE_MESH_LOCAL_PEER_ADD);
   msg.tunnel_id = htonl (tunnel->tid);
   msg.peer = *peer;
-  send_packet (tunnel->mesh, &msg.header);
+  send_packet (tunnel->mesh, &msg.header, tunnel);
 
   return;
 }
@@ -1409,7 +1828,7 @@ GNUNET_MESH_peer_request_connect_del (struct GNUNET_MESH_Tunnel *tunnel,
   msg.header.type = htons (GNUNET_MESSAGE_TYPE_MESH_LOCAL_PEER_DEL);
   msg.tunnel_id = htonl (tunnel->tid);
   memcpy (&msg.peer, peer, sizeof (struct GNUNET_PeerIdentity));
-  send_packet (tunnel->mesh, &msg.header);
+  send_packet (tunnel->mesh, &msg.header, tunnel);
 }
 
 
@@ -1429,25 +1848,108 @@ GNUNET_MESH_peer_request_connect_by_type (struct GNUNET_MESH_Tunnel *tunnel,
 
   GNUNET_array_append (tunnel->apps, tunnel->napps, app_type);
 
+  LOG (GNUNET_ERROR_TYPE_DEBUG, "* CONNECT BY TYPE *\n");
   msg.header.size = htons (sizeof (struct GNUNET_MESH_ConnectPeerByType));
   msg.header.type = htons (GNUNET_MESSAGE_TYPE_MESH_LOCAL_PEER_ADD_BY_TYPE);
   msg.tunnel_id = htonl (tunnel->tid);
   msg.type = htonl (app_type);
-  send_packet (tunnel->mesh, &msg.header);
+  send_packet (tunnel->mesh, &msg.header, tunnel);
+}
+
+
+/**
+ * Request that the mesh should try to connect to a peer matching the
+ * description given in the service string.
+ * 
+ * FIXME: allow multiple? how to deal with reconnect?
+ *
+ * @param tunnel handle to existing tunnel
+ * @param description string describing the destination node requirements
+ */
+void
+GNUNET_MESH_peer_request_connect_by_string (struct GNUNET_MESH_Tunnel *tunnel,
+                                            const char *description)
+{
+  struct GNUNET_MESH_ConnectPeerByString *m;
+  size_t len;
+  size_t msgsize;
+
+  len = strlen (description);
+  msgsize = sizeof(struct GNUNET_MESH_ConnectPeerByString) + len;
+  GNUNET_assert (UINT16_MAX > msgsize);
+  {
+    char buffer[msgsize];
+
+    m = (struct GNUNET_MESH_ConnectPeerByString *) buffer;
+    m->header.size = htons (msgsize);
+    m->header.type = htons (GNUNET_MESSAGE_TYPE_MESH_LOCAL_PEER_ADD_BY_STRING);
+    m->tunnel_id = htonl (tunnel->tid);
+    memcpy(&m[1], description, len);
+
+    send_packet (tunnel->mesh, &m->header, tunnel);
+  }
+}
+
+
+/**
+ * Request that the given peer isn't added to this tunnel in calls to
+ * connect_by_* calls, (due to misbehaviour, bad performance, ...).
+ *
+ * @param tunnel handle to existing tunnel.
+ * @param peer peer identity of the peer which should be blacklisted
+ *                  for the tunnel.
+ */
+void
+GNUNET_MESH_peer_blacklist (struct GNUNET_MESH_Tunnel *tunnel,
+                            const struct GNUNET_PeerIdentity *peer)
+{
+  struct GNUNET_MESH_PeerControl msg;
+
+  msg.header.size = htons (sizeof (struct GNUNET_MESH_PeerControl));
+  msg.header.type = htons (GNUNET_MESSAGE_TYPE_MESH_LOCAL_PEER_BLACKLIST);
+  msg.tunnel_id = htonl (tunnel->tid);
+  msg.peer = *peer;
+  send_packet (tunnel->mesh, &msg.header, tunnel);
+
+  return;
+}
+
+
+/**
+ * Request that the given peer isn't blacklisted anymore from this tunnel,
+ * and therefore can be added in future calls to connect_by_*.
+ * The peer must have been previously blacklisted for this tunnel.
+ *
+ * @param tunnel handle to existing tunnel.
+ * @param peer peer identity of the peer which shouldn't be blacklisted
+ *                  for the tunnel anymore.
+ */
+void
+GNUNET_MESH_peer_unblacklist (struct GNUNET_MESH_Tunnel *tunnel,
+                              const struct GNUNET_PeerIdentity *peer)
+{
+  struct GNUNET_MESH_PeerControl msg;
+
+  msg.header.size = htons (sizeof (struct GNUNET_MESH_PeerControl));
+  msg.header.type = htons (GNUNET_MESSAGE_TYPE_MESH_LOCAL_PEER_UNBLACKLIST);
+  msg.tunnel_id = htonl (tunnel->tid);
+  msg.peer = *peer;
+  send_packet (tunnel->mesh, &msg.header, tunnel);
+
+  return;
 }
 
 
 /**
  * Ask the mesh to call "notify" once it is ready to transmit the
- * given number of bytes to the specified "target".  If we are not yet
- * connected to the specified peer, a call to this function will cause
- * us to try to establish a connection.
+ * given number of bytes to the specified tunnel or target.
+ * Only one call can be active at any time, to issue another request,
+ * wait for the callback or cancel the current request.
  *
  * @param tunnel tunnel to use for transmission
  * @param cork is corking allowed for this transmission?
- * @param priority how important is the message?
  * @param maxdelay how long can the message wait?
- * @param target destination for the message,
+ * @param target destination for the message
  *               NULL for multicast to all tunnel targets
  * @param notify_size how many bytes of buffer space does notify want?
  * @param notify function to call when buffer space is available;
@@ -1461,7 +1963,6 @@ GNUNET_MESH_peer_request_connect_by_type (struct GNUNET_MESH_Tunnel *tunnel,
  */
 struct GNUNET_MESH_TransmitHandle *
 GNUNET_MESH_notify_transmit_ready (struct GNUNET_MESH_Tunnel *tunnel, int cork,
-                                   uint32_t priority,
                                    struct GNUNET_TIME_Relative maxdelay,
                                    const struct GNUNET_PeerIdentity *target,
                                    size_t notify_size,
@@ -1469,56 +1970,19 @@ GNUNET_MESH_notify_transmit_ready (struct GNUNET_MESH_Tunnel *tunnel, int cork,
                                    void *notify_cls)
 {
   struct GNUNET_MESH_TransmitHandle *th;
-  struct GNUNET_MESH_TransmitHandle *least_priority_th;
-  uint32_t least_priority;
   size_t overhead;
 
-#if MESH_API_DEBUG
-  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-              "mesh: mesh notify transmit ready called\n");
+  GNUNET_assert (NULL != tunnel);
+  LOG (GNUNET_ERROR_TYPE_DEBUG, "mesh notify transmit ready called\n");
   if (NULL != target)
-    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "mesh:     target %s\n",
-                GNUNET_i2s (target));
+    LOG (GNUNET_ERROR_TYPE_DEBUG, "    target %s\n", GNUNET_i2s (target));
   else
-    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "mesh:     target multicast\n");
-#endif
+    LOG (GNUNET_ERROR_TYPE_DEBUG, "    target multicast\n");
+  LOG (GNUNET_ERROR_TYPE_DEBUG, "    payload size %u\n", notify_size);
   GNUNET_assert (NULL != notify);
-  if (tunnel->mesh->npackets >= tunnel->mesh->max_queue_size &&
-      tunnel->npackets > 0)
-  {
-    /* queue full */
-    if (0 == priority)
-      return NULL;
-    th = tunnel->mesh->th_tail;
-    least_priority = priority;
-    least_priority_th = NULL;
-    while (NULL != th)
-    {
-      if (th->priority < least_priority && th->tunnel->npackets > 1)
-      {
-        least_priority_th = th;
-        least_priority = th->priority;
-      }
-      th = th->prev;
-    }
-    if (NULL == least_priority_th)
-      return NULL;
-    /* Can't be a control message */
-    GNUNET_assert (NULL != least_priority_th->notify);
-    least_priority_th->notify (notify_cls, 0, NULL);
-    least_priority_th->tunnel->npackets--;
-    tunnel->mesh->npackets--;
-    GNUNET_CONTAINER_DLL_remove (tunnel->mesh->th_head, tunnel->mesh->th_tail,
-                                 least_priority_th);
-    if (GNUNET_SCHEDULER_NO_TASK != least_priority_th->timeout_task)
-      GNUNET_SCHEDULER_cancel (least_priority_th->timeout_task);
-    GNUNET_free (least_priority_th);
-  }
-  tunnel->npackets++;
-  tunnel->mesh->npackets++;
+  GNUNET_assert (0 == tunnel->packet_size); // Only one data packet allowed
   th = GNUNET_malloc (sizeof (struct GNUNET_MESH_TransmitHandle));
   th->tunnel = tunnel;
-  th->priority = priority;
   th->timeout = GNUNET_TIME_relative_to_absolute (maxdelay);
   th->target = GNUNET_PEER_intern (target);
   if (tunnel->tid >= GNUNET_MESH_LOCAL_TUNNEL_ID_SERV)
@@ -1527,12 +1991,17 @@ GNUNET_MESH_notify_transmit_ready (struct GNUNET_MESH_Tunnel *tunnel, int cork,
     overhead = sizeof (struct GNUNET_MESH_Multicast);
   else
     overhead = sizeof (struct GNUNET_MESH_Unicast);
-  th->size = notify_size + overhead;
+  tunnel->packet_size = th->size = notify_size + overhead;
+  LOG (GNUNET_ERROR_TYPE_DEBUG, "    total size %u\n", th->size);
   th->notify = notify;
   th->notify_cls = notify_cls;
   add_to_queue (tunnel->mesh, th);
   if (NULL != tunnel->mesh->th)
     return th;
+  if (GNUNET_NO == PID_OVERFLOW(tunnel->pid, tunnel->max_pid) &&
+      tunnel->max_pid <= tunnel->pid)
+    return th;
+  LOG (GNUNET_ERROR_TYPE_DEBUG, "    call notify tmt rdy\n");
   tunnel->mesh->th =
       GNUNET_CLIENT_notify_transmit_ready (tunnel->mesh->client, th->size,
                                            GNUNET_TIME_UNIT_FOREVER_REL,
@@ -1552,12 +2021,13 @@ GNUNET_MESH_notify_transmit_ready_cancel (struct GNUNET_MESH_TransmitHandle *th)
 {
   struct GNUNET_MESH_Handle *mesh;
 
+  th->tunnel->packet_size = 0;
   mesh = th->tunnel->mesh;
   if (th->timeout_task != GNUNET_SCHEDULER_NO_TASK)
     GNUNET_SCHEDULER_cancel (th->timeout_task);
   GNUNET_CONTAINER_DLL_remove (mesh->th_head, mesh->th_tail, th);
   GNUNET_free (th);
-  if ((NULL == mesh->th_head) && (NULL != mesh->th))
+  if ((0 == message_ready_size (mesh)) && (NULL != mesh->th))
   {
     /* queue empty, no point in asking for transmission */
     GNUNET_CLIENT_notify_transmit_ready_cancel (mesh->th);
@@ -1585,9 +2055,3 @@ GNUNET_MESH_tunnel_get_data (struct GNUNET_MESH_Tunnel *tunnel)
 }
 
 
-#if 0                           /* keep Emacsens' auto-indent happy */
-{
-#endif
-#ifdef __cplusplus
-}
-#endif