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

diff --git a/src/mesh/mesh_api.c b/src/mesh/mesh_api.c
index 8117b9354..6bdc25695 100644
--- a/src/mesh/mesh_api.c
+++ b/src/mesh/mesh_api.c
@@ -21,8 +21,8 @@
  * @author Bartlomiej Polot
  *
  * STRUCTURE:
- * - CONSTANTS
  * - DATA STRUCTURES
+ * - DECLARATIONS
  * - AUXILIARY FUNCTIONS
  * - RECEIVE HANDLERS
  * - SEND FUNCTIONS
@@ -42,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     ****************************/
 /******************************************************************************/
@@ -307,9 +298,9 @@ 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
@@ -340,10 +331,68 @@ struct GNUNET_MESH_Tunnel
 };
 
 
+/******************************************************************************/
+/***********************         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
@@ -394,6 +443,7 @@ create_tunnel (struct GNUNET_MESH_Handle *h, MESH_TunnelNumber tid)
   {
     t->tid = tid;
   }
+  t->max_pid = 1;
   return t;
 }
 
@@ -453,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 */
@@ -464,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;
@@ -554,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
  */
@@ -565,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;
   }
@@ -621,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
@@ -694,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)
   {
@@ -808,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);
@@ -1042,6 +1118,7 @@ process_incoming_data (struct GNUNET_MESH_Handle *h,
       {
         LOG (GNUNET_ERROR_TYPE_DEBUG,
              "callback completed successfully\n");
+        send_ack (h, t);
       }
     }
   }
@@ -1077,8 +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);
+  }
 }
 
 
@@ -1095,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 */
@@ -1129,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");
+  }
 }
 
 
@@ -1157,24 +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, "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;
-  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;
@@ -1183,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);
-        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);
           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));
@@ -1206,16 +1315,16 @@ 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);
           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.pid = 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));
@@ -1230,58 +1339,64 @@ 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);
           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 %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;
     }
     if (th->timeout_task != GNUNET_SCHEDULER_NO_TASK)
       GNUNET_SCHEDULER_cancel (th->timeout_task);
-    if (NULL != th->notify)
-    {
-      th->tunnel->npackets--;
-    }
     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, "  total size: %u\n", tsize);
-  if (NULL != (th = h->th_head))
+  h->th = NULL;
+  size = message_ready_size (h);
+  if (0 != size)
   {
-    LOG (GNUNET_ERROR_TYPE_DEBUG, "  next size: %u\n", th->size);
-    if (NULL == h->th)
-      h->th =
-          GNUNET_CLIENT_notify_transmit_ready (h->client, th->size,
-                                               GNUNET_TIME_UNIT_FOREVER_REL,
-                                               GNUNET_YES, &send_callback, h);
+    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;
 }
 
@@ -1352,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;
@@ -1415,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))
     {
@@ -1546,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);
@@ -1862,9 +1978,9 @@ 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->npackets);
-  tunnel->npackets++;
+  GNUNET_assert (0 == tunnel->packet_size); // Only one data packet allowed
   th = GNUNET_malloc (sizeof (struct GNUNET_MESH_TransmitHandle));
   th->tunnel = tunnel;
   th->timeout = GNUNET_TIME_relative_to_absolute (maxdelay);
@@ -1875,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,
@@ -1900,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);