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

diff --git a/src/mesh/mesh_api.c b/src/mesh/mesh_api.c
index aee4c2a7b..6bdc25695 100644
--- a/src/mesh/mesh_api.c
+++ b/src/mesh/mesh_api.c
@@ -21,7 +21,6 @@
  * @author Bartlomiej Polot
  *
  * STRUCTURE:
- * - CONSTANTS
  * - DATA STRUCTURES
  * - DECLARATIONS
  * - AUXILIARY FUNCTIONS
@@ -43,15 +42,6 @@
 #define LOG(kind,...) GNUNET_log_from (kind, "mesh-api",__VA_ARGS__)
 
 
-/******************************************************************************/
-/************************        CONSTANTS         ****************************/
-/******************************************************************************/
-
-#define HIGH_PID 0xFFFF0000
-#define LOW_PID 0x0000FFFF
-
-#define PID_OVERFLOW(pid, max) (pid > HIGH_PID && max < LOW_PID)
-
 /******************************************************************************/
 /************************      DATA STRUCTURES     ****************************/
 /******************************************************************************/
@@ -363,6 +353,46 @@ 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
@@ -413,6 +443,7 @@ create_tunnel (struct GNUNET_MESH_Handle *h, MESH_TunnelNumber tid)
   {
     t->tid = tid;
   }
+  t->max_pid = 1;
   return t;
 }
 
@@ -472,7 +503,7 @@ destroy_tunnel (struct GNUNET_MESH_Tunnel *t, int call_cleaner)
       continue;
     /* Clients should have aborted their requests already.
      * Management traffic should be ok, as clients can't cancel that */
-    GNUNET_break (NULL == th->notify);
+    GNUNET_break (GNUNET_NO == th_is_payload(th));
     GNUNET_CONTAINER_DLL_remove (h->th_head, h->th_tail, th);
 
     /* clean up request */
@@ -483,7 +514,7 @@ destroy_tunnel (struct GNUNET_MESH_Tunnel *t, int call_cleaner)
 
   /* if there are no more pending requests with mesh service, cancel active request */
   /* Note: this should be unnecessary... */
-  if ( (NULL == h->th_head) && (NULL != h->th))
+  if ((0 == message_ready_size (h)) && (NULL != h->th))
   {
     GNUNET_CLIENT_notify_transmit_ready_cancel (h->th);
     h->th = NULL;
@@ -573,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
  */
@@ -584,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;
   }
@@ -640,6 +673,28 @@ send_packet (struct GNUNET_MESH_Handle *h,
              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;
+}
+
+
+
 /**
  * Reconnect callback: tries to reconnect again after a failer previous
  * reconnecttion
@@ -713,8 +768,9 @@ do_reconnect (struct GNUNET_MESH_Handle *h)
   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");
 
-  h->in_receive = GNUNET_NO;
   /* disconnect */
   if (NULL != h->th)
   {
@@ -827,6 +883,7 @@ 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);
@@ -1061,6 +1118,7 @@ process_incoming_data (struct GNUNET_MESH_Handle *h,
       {
         LOG (GNUNET_ERROR_TYPE_DEBUG,
              "callback completed successfully\n");
+        send_ack (h, t);
       }
     }
   }
@@ -1096,13 +1154,19 @@ process_ack (struct GNUNET_MESH_Handle *h,
     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);
+  }
 }
 
 
@@ -1119,12 +1183,12 @@ msg_received (void *cls, const struct GNUNET_MessageHeader *msg)
 
   if (msg == NULL)
   {
-    LOG (GNUNET_ERROR_TYPE_DEBUG, "Received NULL msg\n");
+    LOG (GNUNET_ERROR_TYPE_WARNING, "Received NULL msg on %p\n", h);
     reconnect (h);
     return;
   }
-  LOG (GNUNET_ERROR_TYPE_DEBUG, "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 */
@@ -1153,12 +1217,20 @@ msg_received (void *cls, const struct GNUNET_MessageHeader *msg)
   default:
     /* We shouldn't get any other packages, log and ignore */
     LOG (GNUNET_ERROR_TYPE_WARNING,
-         "unsolicited message form service (type %d)\n",
+         "unsolicited message form service (type %hu)\n",
          ntohs (msg->type));
   }
   LOG (GNUNET_ERROR_TYPE_DEBUG, "message processed\n");
-  GNUNET_CLIENT_receive (h->client, &msg_received, h,
-                         GNUNET_TIME_UNIT_FOREVER_REL);
+  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");
+  }
 }
 
 
@@ -1188,20 +1260,23 @@ send_callback (void *cls, size_t size, void *buf)
   size_t psize;
 
   LOG (GNUNET_ERROR_TYPE_DEBUG, "Send packet() Buffer %u\n", size);
-  h->th = NULL;
   if ((0 == size) || (NULL == buf))
   {
-    LOG (GNUNET_ERROR_TYPE_DEBUG, "Received NULL callback\n");
+    LOG (GNUNET_ERROR_TYPE_DEBUG, "Received NULL send callback on %p\n", h);
     reconnect (h);
+    h->th = NULL;
     return 0;
   }
   tsize = 0;
+  next = h->th_head;
   while ((NULL != (th = next)) && (size >= th->size))
   {
     t = th->tunnel;
-    if (NULL != th->notify)
+    if (GNUNET_YES == th_is_payload (th))
     {
-      if (t->max_pid < t->pid && ! PID_OVERFLOW (t->pid, t->max_pid)) {
+      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;
@@ -1216,8 +1291,8 @@ 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);
-        LOG (GNUNET_ERROR_TYPE_DEBUG, "  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);
@@ -1240,8 +1315,8 @@ 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);
-        LOG (GNUNET_ERROR_TYPE_DEBUG, "  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);
@@ -1264,8 +1339,8 @@ 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);
-        LOG (GNUNET_ERROR_TYPE_DEBUG, "  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);
@@ -1284,8 +1359,9 @@ send_callback (void *cls, size_t size, void *buf)
     else
     {
       struct GNUNET_MessageHeader *mh = (struct GNUNET_MessageHeader *) &th[1];
-      LOG (GNUNET_ERROR_TYPE_DEBUG, "  mesh traffic, type %u\n",
-             ntohs (mh->type));
+
+      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;
     }
@@ -1299,42 +1375,28 @@ send_callback (void *cls, size_t size, void *buf)
     tsize += psize;
   }
   LOG (GNUNET_ERROR_TYPE_DEBUG, "  total size: %u\n", tsize);
-  if (NULL != h->th_head)
+  h->th = NULL;
+  size = message_ready_size (h);
+  if (0 != size)
   {
-    int request = GNUNET_NO;
-
-    for (th = h->th_head; NULL != th; th = th->next)
-    {
-      struct GNUNET_MESH_Tunnel *t = th->tunnel;
-
-      if (NULL != th->notify || 
-          (t->max_pid >= t->pid || PID_OVERFLOW (t->pid, t->max_pid)))
-      {
-        request = GNUNET_YES;
-        break;
-      }
-    }
-
-    if (GNUNET_YES == request)
-    {
-      LOG (GNUNET_ERROR_TYPE_DEBUG, "  next size: %u\n", th->size);
-      h->th =
-          GNUNET_CLIENT_notify_transmit_ready (h->client, th->size,
-                                               GNUNET_TIME_UNIT_FOREVER_REL,
-                                               GNUNET_YES, &send_callback, h);
-    }
-    else
-    {
-      LOG (GNUNET_ERROR_TYPE_DEBUG, "  nothing left to transmit\n");
-    }
+    LOG (GNUNET_ERROR_TYPE_DEBUG, "  next size: %u\n", size);
+    h->th =
+        GNUNET_CLIENT_notify_transmit_ready (h->client, size,
+                                             GNUNET_TIME_UNIT_FOREVER_REL,
+                                             GNUNET_YES, &send_callback, h);
+  }
+  else
+  {
+    LOG (GNUNET_ERROR_TYPE_DEBUG, "  nothing left to transmit\n");
   }
-  LOG (GNUNET_ERROR_TYPE_DEBUG, "Send packet() END\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;
 }
 
@@ -1405,6 +1467,7 @@ GNUNET_MESH_connect (const struct GNUNET_CONFIGURATION_Handle *cfg, void *cls,
 
   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->new_tunnel = new_tunnel;
   h->cleaner = cleaner;
@@ -1468,7 +1531,7 @@ GNUNET_MESH_disconnect (struct GNUNET_MESH_Handle *handle)
     /* Make sure it is an allowed packet (everything else should have been
      * already canceled).
      */
-    GNUNET_break (NULL == th->notify);
+    GNUNET_break (GNUNET_NO == th_is_payload (th));
     msg = (struct GNUNET_MessageHeader *) &th[1];
     switch (ntohs(msg->type))
     {
@@ -1599,7 +1662,7 @@ GNUNET_MESH_tunnel_destroy (struct GNUNET_MESH_Tunnel *tunnel)
     {
       aux = th->next;
       /* FIXME call the handler? */
-      if (NULL != th->notify)
+      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);
@@ -1915,6 +1978,7 @@ GNUNET_MESH_notify_transmit_ready (struct GNUNET_MESH_Tunnel *tunnel, int cork,
     LOG (GNUNET_ERROR_TYPE_DEBUG, "    target %s\n", GNUNET_i2s (target));
   else
     LOG (GNUNET_ERROR_TYPE_DEBUG, "    target multicast\n");
+  LOG (GNUNET_ERROR_TYPE_DEBUG, "    payload size %u\n", notify_size);
   GNUNET_assert (NULL != notify);
   GNUNET_assert (0 == tunnel->packet_size); // Only one data packet allowed
   th = GNUNET_malloc (sizeof (struct GNUNET_MESH_TransmitHandle));
@@ -1928,11 +1992,16 @@ GNUNET_MESH_notify_transmit_ready (struct GNUNET_MESH_Tunnel *tunnel, int cork,
   else
     overhead = sizeof (struct GNUNET_MESH_Unicast);
   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,
@@ -1952,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);